Core: add schema id and schemas to table metadata #2096

yyanyy · 2021-01-16T02:01:00Z

This PR adds current-schema-id and schemas to table metadata. It also introduces a wrapper around schema to associate table schema with id.
The reason to not add ID directly into Schema is that currently schema creation is widely used as a convenient method for a lot of actions that don't involve a "real" table schema.

Next steps:

adds schema-id to snapshot interface and populate
use history entries and schemas to look up the right schema in time travel queries; this may mean to add schemas() in Table API
spec update
add schema id to historyTable (will be mentioned later)

Open questions:

Current approach writes the newly introduced fields to JSON by default even in v1, and there could be forward/backward compatibility concern with the current approach: if a new writer writes (with ID 0) and then update (with ID 1) schema, metadata will store both schema ID 0 and 1, and default ID will be 1. Then an old writer reads and writes the metadata for whatever change, which drops schema 0 in metadata. Then when a new writer picks up the metadata again, the original schema 0 is gone, and 1 is replaced with ID 0. This could result in schema ID consistency issue among different writers.
- Since ID is introduced in this PR, there is no metadata table that exposes these inconsistent schema IDs, so we may not have this problem for now. However when we start to add schema ID to historyTable metadata table, at different time ID 0 could mean different things in this history table. We could potentially workaround this by only exposing schemaID field in historyTable only for v2 tables, or mention this caveat on spec.
- Alternatively we can expose these two fields only in v2 table, and time travel queries in v1 always rely on looking at old table metadata files as implemented in Use schema at the time of the snapshot when reading a snapshot. #1508. This could mean in future any new changes that may depend on schema ID cannot be introduced in v1.
Do we want to add a last-assigned-schema-id to table metadata? My answer would be yes, for a similar reason mentioned in this comment
Do we want to assign special schema IDs to metadata tables, to avoid potential collision on schema IDs? Currently I don't think this is going to be a problem since different data tables could also have this collision problem, and plan to assign 0 for all metadata tables' schema for now, but wanted to bring it up as this may not be easy to fix if we want to revisit this decision.
Do we want to add snapshot-id to only history entries, or Snapshot interface? I think former is enough for fixing the time travel query problem, but we might want the latter in case we need this information later.
I think currently when replacing a table, earlier history entries/snapshotLog will be reset to empty (second to last argument in here). Is this expected? do we want to fix this as a separate issue?

jackye1995 · 2021-01-16T19:49:42Z

core/src/main/java/org/apache/iceberg/TableMetadataParser.java

@@ -242,6 +255,7 @@ public static TableMetadata read(FileIO io, InputFile file) {
    }
  }

+  @SuppressWarnings("checkstyle:CyclomaticComplexity")


I remember this also shows up when adding primary key #2010 , I think we should probably find some way to refactor the code a bit.

Yeah I imagine there would be some conflict when either of the PRs got merged. I think the main problem with refactoring this is that a lot of the things it parsed should be parsed together/validate each other; e.g. parse of schema array has to come together with current schema id since both will either present at the same time, or derived from a different field, and same goes for sort order/partition spec. Since Java only allows returning a single object per method, we either have to refactor the code a bit and ensure the logic flow doesn't change and is readable, or let the new methods to return things like tuple<Integer, List<Schema>> that I'm not sure which way is cleaner.

I think we could have a method that parses schemas and a separate one that returns the current schema ID. That said, I'm fine with suppressing the complexity warning for now.

jackye1995 · 2021-01-16T19:53:23Z

api/src/main/java/org/apache/iceberg/VersionedSchema.java

+/**
+ * A wrapper around Schema to include its schema ID
+ */
+public class VersionedSchema implements Serializable {


The biggest question I have for this PR is whether we should have this wrapped class for adding schema ID, or we should just add schema ID to the Schema class itself, maybe as an optional Integer. Because Iceberg have its own jackson parser for serdes, it is not hard to ignore the schemaId field or set a default for it for backwards compatibility. What is the benefit of this VersionedSchema approach?

I mentioned this briefly in the PR description:

The reason to not add ID directly into Schema is that currently schema creation is widely used as a convenient method for a lot of actions that don't involve a "real" table schema.

The main concern I had with adding schema ID is that it might be confusing to tell from the code base when a schema should have ID and when not. A schema only needs to have ID when it's really a table schema strictly associated with table metadata, but outside of that context, the ID is not useful in any "real" operation of the schema, especially considering that it's often used for conveniently mutate fields for things like projection. Maybe we should rename VersionedSchema to something like TableSchema to better reflect this?

I think I agree with @jackye1995 here. We could add the ID to Schema and just have it use either null or 0 if there is no schema ID. I think we can use 0.

If there is no schema ID, then there will not be a schemas list, and all of the snapshots would not have a schema ID. So if an older writer updates a table, then we can expect the current schema ID to be dropped, older schemas to be dropped, snapshot schema IDs to be dropped, etc. So it is consistent with having just one schema. That should be safe for v1 because we won't leave any dangling IDs that could reference the wrong schema later.

My comment above is thinking through your question 1 in the description. I think it is okay to do this as you have it here.

rdblue · 2021-01-30T21:53:49Z

api/src/main/java/org/apache/iceberg/Schema.java

+  @Override
+  public int hashCode() {
+    return Objects.hash(struct, aliasToId);
+  }


Why add equals and hashCode to Schema? The reason why we didn't already have this implemented is that you can compare the type using schema.asStruct().equals(...). And we don't want to define whether two schemas are "equal" based on the aliases because the same schema can have different original aliases.

In the new change I introduced, we can reuse the old schema ID during schema update, if the exact same schema already exists in the schema list. I think in this case we may need to confirm both schemas' aliases should be the same? Or do you think there shouldn't be a case when schema are the same while aliases are not?

I don't think that should use aliases. Aliases are populated when converting from a file schema to Iceberg, so that we can look up the file's column names from IDs and vice versa. That information is used by integration and not by Iceberg itself. The aliases are not serialized when writing a schema into Iceberg metadata.

For the purposes of maintaining the schema list, we can ignore aliases and just use schema.asStruct().

core/src/main/java/org/apache/iceberg/SchemaParser.java

rdblue · 2021-01-30T22:05:34Z

core/src/main/java/org/apache/iceberg/TableMetadataParser.java

@@ -161,8 +163,19 @@ private static void toJson(TableMetadata metadata, JsonGenerator generator) thro
    generator.writeNumberField(LAST_UPDATED_MILLIS, metadata.lastUpdatedMillis());
    generator.writeNumberField(LAST_COLUMN_ID, metadata.lastColumnId());

-    generator.writeFieldName(SCHEMA);
-    SchemaParser.toJson(metadata.schema(), generator);
+    // for older readers, continue writing the current schema as "schema"


Might want to note that this is done until v2 because support for schemas and current-schema-id is required in v2 and later.

rdblue · 2021-01-30T22:40:52Z

Do we want to add a last-assigned-schema-id to table metadata?

My initial thought was no because the schema IDs are completely local to a metadata file. We can reassign all IDs from one file to the next and as long as they are internally consistent we are okay. The IDs are not embedded in separate files that might get improperly used.

But, the Nessie comment you pointed to is a good one to think about. I think it is still okay because Nessie merges would either be fast-forward, or would re-apply changes and create entirely new metadata files. So the internal consistency argument still holds.

Do we want to assign special schema IDs to metadata tables, to avoid potential collision on schema IDs?

I don't think so.

Do we want to add snapshot-id to only history entries, or Snapshot interface?

I would add it to Snapshot, not just History. Snapshot is what will be used for time travel queries. I would avoid adding too much metadata to history.

I think currently when replacing a table, earlier history entries/snapshotLog will be reset to empty (second to last argument in here). Is this expected? do we want to fix this as a separate issue?

I think we should follow up and keep the history. I think the reason for this is because we previously didn't have any compatibility across schemas (IDs were completely reassigned, so time travel would be incorrect). But that's fixed now.

yyanyy · 2021-02-02T03:10:34Z

Do we want to add a last-assigned-schema-id to table metadata?

My initial thought was no because the schema IDs are completely local to a metadata file. We can reassign all IDs from one file to the next and as long as they are internally consistent we are okay. The IDs are not embedded in separate files that might get improperly used.

But, the Nessie comment you pointed to is a good one to think about. I think it is still okay because Nessie merges would either be fast-forward, or would re-apply changes and create entirely new metadata files. So the internal consistency argument still holds.

Do we want to assign special schema IDs to metadata tables, to avoid potential collision on schema IDs?

I don't think so.

Do we want to add snapshot-id to only history entries, or Snapshot interface?

I would add it to Snapshot, not just History. Snapshot is what will be used for time travel queries. I would avoid adding too much metadata to history.

I think currently when replacing a table, earlier history entries/snapshotLog will be reset to empty (second to last argument in here). Is this expected? do we want to fix this as a separate issue?

I think we should follow up and keep the history. I think the reason for this is because we previously didn't have any compatibility across schemas (IDs were completely reassigned, so time travel would be incorrect). But that's fixed now.

Thank you for the review and the input! I'll create a separate issue to mention the history problem and link here. Do you have any suggestion on if we want to add schema id to the history table?

rdblue · 2021-02-03T00:06:18Z

Do you have any suggestion on if we want to add schema id to the history table?

I don't think that we want to add it to history. The snapshot log should refer to snapshots by ID and we can get the schema easily enough from there. For the actual metadata table, we don't want to expose schema ID because that's something that users should ideally never need to know much about. We've avoided adding IDs in other similar situations. For example, you can rewrite manifests for a specific partition spec, by ID. But we didn't include that in the stored procedure because we haven't exposed the partition spec ID anywhere to SQL users.

jackye1995

Looks good to me from code, I am testing locally with this branch to see if there are any other potential issues, will reply back after that.

jackye1995 · 2021-02-04T18:41:51Z

api/src/main/java/org/apache/iceberg/types/TypeUtil.java

@@ -157,6 +157,21 @@ public static Schema assignFreshIds(Schema schema, NextID nextId) {
        .fields());
  }

+  /**
+   * Assigns fresh ids from the {@link NextID nextId function} for all fields in a schema.


nit: function should be outside the bracket

Hmm actually I think either way may make sense? We either link nextId function or nextId to NextId interface, and I think both are valid interpretations? I'm also a bit hesitant to change since I copied this from another method, and there are a few others that use this style.

jackye1995 · 2021-02-04T19:21:01Z

core/src/main/java/org/apache/iceberg/SchemaParser.java

@@ -134,6 +143,10 @@ static void toJson(Type type, JsonGenerator generator) throws IOException {
    }
  }

+  public static void toJsonWithVersion(Schema schema, JsonGenerator generator) throws IOException {


nit: should this be called toJsonWithId, just to match the fromJsonWithId below?

Good point, will update

jackye1995 · 2021-02-07T21:42:49Z

Works fine for me so far, tested a few operations among a few clusters of different versions.

One behavior I see is that, if an old v1 writer is somehow still used to create a new table version after v2 writer is used, all schema history will be dropped, so new schema written by v2 writer will have schema id starting from 0 again, which can break time travel query.

In theory, once a table is written by a v2 writer, it should never be touched by a v1 writer, but we cannot prevent people from doing that in real life, especially when the organization is large and people use different versions of the same software all the time. So I feel we still need something like a last-assigned-schema-id, so that the schemas don't accidentally share the same id and we can backfill the earlier schema and their IDs if necessary. But last-assigned-schema-id field should not as a part of the table metadata field because it will be dropped by an old writer. One way to go is to maybe put it in table properties, but there might be better ways. What do you think?

yyanyy · 2021-02-09T01:58:19Z

Works fine for me so far, tested a few operations among a few clusters of different versions.

One behavior I see is that, if an old v1 writer is somehow still used to create a new table version after v2 writer is used, all schema history will be dropped, so new schema written by v2 writer will have schema id starting from 0 again, which can break time travel query.

In theory, once a table is written by a v2 writer, it should never be touched by a v1 writer, but we cannot prevent people from doing that in real life, especially when the organization is large and people use different versions of the same software all the time. So I feel we still need something like a last-assigned-schema-id, so that the schemas don't accidentally share the same id and we can backfill the earlier schema and their IDs if necessary. But last-assigned-schema-id field should not as a part of the table metadata field because it will be dropped by an old writer. One way to go is to maybe put it in table properties, but there might be better ways. What do you think?

Thanks for putting efforts into testing this!

I think the problem you described seem to be similar to what I had in my question 1, and I think we might be fine with that since the IDs within the metadata file should always be consistent, and we don't expose them in metadata table as Ryan mentioned above. I think it wouldn't break time travel query, as after the old writer writes the data, the schema ID with the snapshot got dropped, in which case we will likely need to rely on #1508 for time travel?

jackye1995 · 2021-02-09T05:04:00Z

I think the problem you described seem to be similar to what I had in my question 1, and I think we might be fine with that since the IDs within the metadata file should always be consistent, and we don't expose them in metadata table as Ryan mentioned above. I think it wouldn't break time travel query, as after the old writer writes the data, the schema ID with the snapshot got dropped, in which case we will likely need to rely on #1508 for time travel?

I see, that makes sense.

rdblue · 2021-02-13T21:11:59Z

api/src/main/java/org/apache/iceberg/Schema.java

@@ -54,6 +58,7 @@
  private transient Map<Integer, String> idToName = null;

  public Schema(List<NestedField> columns, Map<String, Integer> aliases) {
+    this.schemaId = 0;


I think we should add a constant for the default schema ID. That way we can reference it in other places when we fill in the default.

rdblue · 2021-02-13T21:19:35Z

core/src/main/java/org/apache/iceberg/SchemaParser.java

@@ -134,6 +143,10 @@ static void toJson(Type type, JsonGenerator generator) throws IOException {
    }
  }

+  public static void toJsonWithId(Schema schema, JsonGenerator generator) throws IOException {


This is only called in TableMetadataParser and I don't see why there should be two toJson methods. It isn't a problem for the v1 metadata to be written with the schema ID because extra fields are ignored. So I think that this method isn't needed and toJson should always pass the ID to the struct writer.

rdblue · 2021-02-13T21:20:50Z

core/src/main/java/org/apache/iceberg/SchemaParser.java

+    toJson(struct, null, generator);
+  }
+
+  static void toJson(Types.StructType struct, Integer schemaId, JsonGenerator generator) throws IOException {


Looks like the struct toJson methods aren't called outside of this file. What do you think about making them private?

rdblue · 2021-02-13T21:31:45Z

core/src/main/java/org/apache/iceberg/SchemaParser.java

@@ -253,4 +266,15 @@ public static Schema fromJson(String json) {
      }
    });
  }
+
+  public static Schema fromJsonWithId(int schemaId, JsonNode json) {


I'd prefer if all of the parsing methods returned the correct ID instead of having some that use a version that ignores the ID.

I think you chose to use two methods because there are some situations where the ID wasn't written and you don't want the schema to have the wrong ID. But, because the fromJson method uses the Schema constructor that does not pass the ID, the schemas will have the default ID, 0, anyway.

As long as the schemas have an ID, I think there is no need to have multiple parser methods, both with and without ID. Also, we need to make sure that the IDs are consistent when reading the schema from files that were written before IDs. That should just be manifest files, where the schema is embedded in the header.

For manifests, we need to add the schema ID when writing (https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/ManifestWriter.java#L190). And when reading, we will need to update how the spec is constructed. If the spec is loaded by ID then it's fine. Otherwise we should load the schema by ID to parse the spec, and if that doesn't work we should parse both the schema and the spec (https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/ManifestReader.java#L115-L120). I think we should probably also try to take the parsed schema and find the correct one from table metadata so that we don't have an incorrect ID anywhere.

I have removed fromJsonWithId and update fromJson to use the constructor with schema Id, if Id present in Json blob, or default to constructor that doesn't require Id. Please let me know if it is expected!

Regarding persisting schema ID in manifest, since schema and partition-spec in both v1 and v2 are required in manifest metadata, I think it is unlikely that we will miss schema itself but has schema ID for lookup? And I think in manifest reader, schema is only used for constructing partition spec, so that having a correct schema ID might not be super necessary.

Today there are a lot of places that rely on toJson/fromJson for schema, e.g. Avro metadata, Hive table properties, scan tasks. In this PR, except for table metadata parser, those are all using the toJson version that do not write schemaId at all, which means that when the strings are read back to Schema object they will always have id of 0; and this goes back to the persisting schema ID question above, that we do not persist id in the manifest header, so we are not persisting an incorrect ID, but instead constructed a schema without ID that default to be 0 like everywhere else.

My original thinking was that schema Id should always be ignored/considered inaccurate unless we directly get it from table metadata, as after all kinds of projections it might be hard to track if the ID is correct. And the ID is only used for lookup, so if we have the schema struct, it's fine to have incorrect ID since we are unlikely to use it. But I might be overcomplicating the problem. Do you think we are able to guarantee the ID to be right everywhere, and do you have comment about the current state of toJson variations (that only the one used by metadata parser writes out ID)?

rdblue · 2021-02-13T21:33:10Z