You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "syun64 (via GitHub)" <gi...@apache.org> on 2023/03/01 23:01:45 UTC
[GitHub] [iceberg] syun64 opened a new pull request, #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
syun64 opened a new pull request, #6980:
URL: https://github.com/apache/iceberg/pull/6980
fixes #6978
Will backport the solution to 3.2 and 3.1 if this is approved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123627567
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
Review Comment:
Sorry can we also change the test name to capture the scenario better, maybe testFilesTableTimeTravelWithSchemaEvolution . I know its a bit long, couldnt think anything shorter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] syun64 commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "syun64 (via GitHub)" <gi...@apache.org>.
syun64 commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123664500
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
Review Comment:
Haha, hmmm - I'd argue that the first one is more descriptive of the problem and better describes the motivation in having the specific test, even though it's longer... testFilesTimeTravelAddedColumn is a lot more literal
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1122466917
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:
##########
@@ -165,7 +166,11 @@ private Schema snapshotSchema() {
@Override
public StructType schema() {
if (lazyTableSchema == null) {
- this.lazyTableSchema = SparkSchemaUtil.convert(snapshotSchema());
+ if (icebergTable instanceof BaseMetadataTable) {
+ this.lazyTableSchema = SparkSchemaUtil.convert(icebergTable.schema());
Review Comment:
So this seems to disable time-travel using the right schema for Metadata Tables. I think we should fix in long run, but maybe this is ok for a quick fix for now.
How about, we can put this logic in SnapshotUtil.schemaFor(). If table is BaseMetadataTable, return current the schema. That way, I think we can change just one place in the code? And fix it there later when we want to implement this?
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +489,54 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+ Long olderSnapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s ADD COLUMNS (data2 string)",
+ tableName);
+
+ List<SimpleExtraColumnRecord> recordsB =
+ Lists.newArrayList(new SimpleExtraColumnRecord(1, "b", "c"), new SimpleExtraColumnRecord (2, "b", "c"));
+ spark
+ .createDataset(recordsB, Encoders.bean(SimpleExtraColumnRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+
+ List<Object[]> res1 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, olderSnapshotId );
+
+ Dataset<Row> ds = spark.read().format("iceberg").option("snapshot-id", olderSnapshotId ).load(tableName + ".files");
+ List<Row> res2 = ds.collectAsList();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ List<Object[]> res3 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, currentSnapshotId);
+
+ Dataset<Row> ds2 = spark.read().format("iceberg").option("snapshot-id", currentSnapshotId ).load(tableName + ".files");
Review Comment:
We will actually need to do an assert here on the value, for a meaningful test, see the other examples.
You could look at other examples (testAllFilesPartitioned, which is more complete, but complex):
```
Dataset<Row> actualDataFilesDs =
spark.sql("SELECT * FROM " + tableName + ".files VERSION AS OF %s", tableName, currentSnasphotId);
List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
List<Record> expectedDataFiles =
expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
TestHelpers.assertEqualsSafe(
SparkSchemaUtil.convert(TestHelpers.selectNonDerived(actualDataFilesDs).schema())
.asStruct(),
expectedDataFiles.get(0),
TestHelpers.selectNonDerived(actualDataFilesDs).collectAsList());
```
Or maybe easier, sort the data and check fields individually , like testSnapshotReferencesMetatable
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +489,54 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+ Long olderSnapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s ADD COLUMNS (data2 string)",
+ tableName);
+
+ List<SimpleExtraColumnRecord> recordsB =
+ Lists.newArrayList(new SimpleExtraColumnRecord(1, "b", "c"), new SimpleExtraColumnRecord (2, "b", "c"));
+ spark
+ .createDataset(recordsB, Encoders.bean(SimpleExtraColumnRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+
+ List<Object[]> res1 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, olderSnapshotId );
+
+ Dataset<Row> ds = spark.read().format("iceberg").option("snapshot-id", olderSnapshotId ).load(tableName + ".files");
+ List<Row> res2 = ds.collectAsList();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ List<Object[]> res3 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, currentSnapshotId);
Review Comment:
Let's remove queries that don't matter here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1122467064
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +489,54 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+ Long olderSnapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s ADD COLUMNS (data2 string)",
+ tableName);
+
+ List<SimpleExtraColumnRecord> recordsB =
+ Lists.newArrayList(new SimpleExtraColumnRecord(1, "b", "c"), new SimpleExtraColumnRecord (2, "b", "c"));
+ spark
+ .createDataset(recordsB, Encoders.bean(SimpleExtraColumnRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+
+ List<Object[]> res1 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, olderSnapshotId );
+
+ Dataset<Row> ds = spark.read().format("iceberg").option("snapshot-id", olderSnapshotId ).load(tableName + ".files");
+ List<Row> res2 = ds.collectAsList();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ List<Object[]> res3 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, currentSnapshotId);
Review Comment:
Let's remove queries that don't matter here. Several of these are from my failed attempts to reproduce it, so It wont matter for the test :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho merged pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho merged PR #6980:
URL: https://github.com/apache/iceberg/pull/6980
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] syun64 commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "syun64 (via GitHub)" <gi...@apache.org>.
syun64 commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123644941
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
Review Comment:
Not saying no to good suggestions :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#issuecomment-1452668054
Makes sense, I added it to Iceberg 1.2 milestone. I made a follow up issue to implement this feature for some tables that will be affected , like files table. https://github.com/apache/iceberg/issues/6991.
As other reviewers also commented, so will leave a chance for them to put some comments in.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#issuecomment-1451121645
Also FYI @aokolnychyi , @RussellSpitzer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] syun64 commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "syun64 (via GitHub)" <gi...@apache.org>.
syun64 commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123617309
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+
+ table.updateSchema().addColumn("category", Types.StringType.get()).commit();
+
+ List<Row> newRecords =
+ Lists.newArrayList(RowFactory.create(3, "b", "c"), RowFactory.create(4, "b", "c"));
+
+ StructType newSparkSchema =
+ SparkSchemaUtil.convert(
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "data", Types.StringType.get()),
+ optional(3, "category", Types.StringType.get())));
+
+ spark.createDataFrame(newRecords, newSparkSchema).coalesce(1).writeTo(tableName).append();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ Dataset<Row> actualFilesDs =
+ spark.sql(
+ "SELECT * FROM "
+ + tableName
+ + ".files VERSION AS OF "
+ + currentSnapshotId
+ + " ORDER BY content");
+ List<Row> actualFiles = TestHelpers.selectNonDerived(actualFilesDs).collectAsList();
+ Schema entriesTableSchema = Spark3Util.loadIcebergTable(spark, tableName + ".entries").schema();
+ List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
+ List<Record> expectedFiles =
+ expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
+
+ TestHelpers.assertEqualsSafe(
Review Comment:
I think it's actually size of 1, but I can assert on the files size for sure 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123665168
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
Review Comment:
No problem, up to you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1122495123
##########
spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/SimpleExtraColumnRecord.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
+
+public class SimpleExtraColumnRecord {
Review Comment:
I would also suggest to use existing class, instead of introducing new class for every scenario. Will it work to use:
RowFactory.create()? See TestIcebergSourceTableBase::testSnapshotReadAfterAddColumn for inspiration how we can write a test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123666329
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,65 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesTableTimeTravelWithSchemaEvolution() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+
+ table.updateSchema().addColumn("category", Types.StringType.get()).commit();
+
+ List<Row> newRecords =
+ Lists.newArrayList(RowFactory.create(3, "b", "c"), RowFactory.create(4, "b", "c"));
+
+ StructType newSparkSchema =
+ SparkSchemaUtil.convert(
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "data", Types.StringType.get()),
+ optional(3, "category", Types.StringType.get())));
+
+ spark.createDataFrame(newRecords, newSparkSchema).coalesce(1).writeTo(tableName).append();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ Dataset<Row> actualFilesDs =
+ spark.sql(
+ "SELECT * FROM "
+ + tableName
+ + ".files VERSION AS OF "
+ + currentSnapshotId
+ + " ORDER BY content");
+ List<Row> actualFiles = TestHelpers.selectNonDerived(actualFilesDs).collectAsList();
+ Schema entriesTableSchema = Spark3Util.loadIcebergTable(spark, tableName + ".entries").schema();
+ List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
+ List<Record> expectedFiles =
+ expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
+
+ TestHelpers.assertEqualsSafe(
+ TestHelpers.nonDerivedSchema(actualFilesDs), expectedFiles.get(0), actualFiles.get(0));
+
+ Assert.assertEquals(
Review Comment:
Can you still assert size = 1? I think its not so clear from this assert. If its not 1, then the previous check will be limited.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1124837795
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,70 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesTableTimeTravelWithSchemaEvolution() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+
+ table.updateSchema().addColumn("category", Types.StringType.get()).commit();
+
+ List<Row> newRecords =
+ Lists.newArrayList(RowFactory.create(3, "b", "c"), RowFactory.create(4, "b", "c"));
+
+ StructType newSparkSchema =
+ SparkSchemaUtil.convert(
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "data", Types.StringType.get()),
+ optional(3, "category", Types.StringType.get())));
+
+ spark.createDataFrame(newRecords, newSparkSchema).coalesce(1).writeTo(tableName).append();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ Dataset<Row> actualFilesDs =
+ spark.sql(
+ "SELECT * FROM "
+ + tableName
+ + ".files VERSION AS OF "
+ + currentSnapshotId
+ + " ORDER BY content");
+ List<Row> actualFiles = TestHelpers.selectNonDerived(actualFilesDs).collectAsList();
+ Schema entriesTableSchema = Spark3Util.loadIcebergTable(spark, tableName + ".entries").schema();
+ List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
+ List<Record> expectedFiles =
+ expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
+
+ Assert.assertEquals("actualFiles size should be 1", 2, actualFiles.size());
Review Comment:
Minor: One more fix here for the assert message
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1122500658
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +489,54 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+ Long olderSnapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s ADD COLUMNS (data2 string)",
+ tableName);
+
+ List<SimpleExtraColumnRecord> recordsB =
+ Lists.newArrayList(new SimpleExtraColumnRecord(1, "b", "c"), new SimpleExtraColumnRecord (2, "b", "c"));
+ spark
+ .createDataset(recordsB, Encoders.bean(SimpleExtraColumnRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+
+ List<Object[]> res1 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, olderSnapshotId );
+
+ Dataset<Row> ds = spark.read().format("iceberg").option("snapshot-id", olderSnapshotId ).load(tableName + ".files");
+ List<Row> res2 = ds.collectAsList();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ List<Object[]> res3 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, currentSnapshotId);
+
+ Dataset<Row> ds2 = spark.read().format("iceberg").option("snapshot-id", currentSnapshotId ).load(tableName + ".files");
Review Comment:
Or, on the topic of using's TestIcebergSourceTablesBase::testSnapshotReadAfterAddAndDropColumn mechanism, maybe an easier way is to even use the RowFactory to construct the expected results?
For example, select a few columns you want to check like :
```
Dataset<Row> actual =
spark.sql("SELECT file_path, file_format, record_count FROM " + tableName + ".files VERSION AS OF %s", tableName, currentSnasphotId);
```
and construct expected value RowFactory, with data you get from:
```
TestHelpers.dataFiles(table).stream().map(df -> RowFactory.create(df.path(), df.format(), df.recordCount())).collect(Collectors.toList())
```
up to you, if that is easier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1122500658
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +489,54 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+ Long olderSnapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s ADD COLUMNS (data2 string)",
+ tableName);
+
+ List<SimpleExtraColumnRecord> recordsB =
+ Lists.newArrayList(new SimpleExtraColumnRecord(1, "b", "c"), new SimpleExtraColumnRecord (2, "b", "c"));
+ spark
+ .createDataset(recordsB, Encoders.bean(SimpleExtraColumnRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+
+ List<Object[]> res1 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, olderSnapshotId );
+
+ Dataset<Row> ds = spark.read().format("iceberg").option("snapshot-id", olderSnapshotId ).load(tableName + ".files");
+ List<Row> res2 = ds.collectAsList();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ List<Object[]> res3 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, currentSnapshotId);
+
+ Dataset<Row> ds2 = spark.read().format("iceberg").option("snapshot-id", currentSnapshotId ).load(tableName + ".files");
Review Comment:
Or, on the topic of using's TestIcebergSourceTablesBase::testSnapshotReadAfterAddAndDropColumn mechanism, maybe an easier way is to even use the RowFactory to construct the expected results?
For example, select a few columns you want to check like :
Dataset<Row> actual =
spark.sql("SELECT file_path, file_format, record_count FROM " + tableName + ".files VERSION AS OF %s", tableName, currentSnasphotId);
and construct expected value RowFactory, with data you get from: TestHelpers.dataFiles(table).stream().map(df -> RowFactory.create(df.path, df.format, df.recordCount)).collect...
up to you, if that is easier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123627567
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
Review Comment:
Can we also change the test name to capture the scenario better, maybe testFilesTableTimeTravelWithSchemaEvolution . I know its a bit long, couldnt think anything shorter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#issuecomment-1455278001
Merged, thanks @syun64 . We can track any other discussion in follow up , if any
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1122495123
##########
spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/SimpleExtraColumnRecord.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source;
+
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
+
+public class SimpleExtraColumnRecord {
Review Comment:
I would also suggest to use existing class, instead of introducing new class for every scenario. Will it work to use:
RowFactory.create()? See TestIcebergSourceTableBase::testSnapshotReadAfterAddColumn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] syun64 commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "syun64 (via GitHub)" <gi...@apache.org>.
syun64 commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123194620
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +489,54 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+ Long olderSnapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s ADD COLUMNS (data2 string)",
+ tableName);
+
+ List<SimpleExtraColumnRecord> recordsB =
+ Lists.newArrayList(new SimpleExtraColumnRecord(1, "b", "c"), new SimpleExtraColumnRecord (2, "b", "c"));
+ spark
+ .createDataset(recordsB, Encoders.bean(SimpleExtraColumnRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+
+ List<Object[]> res1 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, olderSnapshotId );
+
+ Dataset<Row> ds = spark.read().format("iceberg").option("snapshot-id", olderSnapshotId ).load(tableName + ".files");
+ List<Row> res2 = ds.collectAsList();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ List<Object[]> res3 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, currentSnapshotId);
+
+ Dataset<Row> ds2 = spark.read().format("iceberg").option("snapshot-id", currentSnapshotId ).load(tableName + ".files");
Review Comment:
Hi @szehon-ho I took your first advice based on testAllFilesPartitioned - may I ask you for a second look through this test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] syun64 commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "syun64 (via GitHub)" <gi...@apache.org>.
syun64 commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123683463
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,65 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesTableTimeTravelWithSchemaEvolution() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+
+ table.updateSchema().addColumn("category", Types.StringType.get()).commit();
+
+ List<Row> newRecords =
+ Lists.newArrayList(RowFactory.create(3, "b", "c"), RowFactory.create(4, "b", "c"));
+
+ StructType newSparkSchema =
+ SparkSchemaUtil.convert(
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "data", Types.StringType.get()),
+ optional(3, "category", Types.StringType.get())));
+
+ spark.createDataFrame(newRecords, newSparkSchema).coalesce(1).writeTo(tableName).append();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ Dataset<Row> actualFilesDs =
+ spark.sql(
+ "SELECT * FROM "
+ + tableName
+ + ".files VERSION AS OF "
+ + currentSnapshotId
+ + " ORDER BY content");
+ List<Row> actualFiles = TestHelpers.selectNonDerived(actualFilesDs).collectAsList();
+ Schema entriesTableSchema = Spark3Util.loadIcebergTable(spark, tableName + ".entries").schema();
+ List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
+ List<Record> expectedFiles =
+ expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
+
+ TestHelpers.assertEqualsSafe(
+ TestHelpers.nonDerivedSchema(actualFilesDs), expectedFiles.get(0), actualFiles.get(0));
+
+ Assert.assertEquals(
Review Comment:
My bad - the size actually ended up being 2 which actually makes much more sense since we have two insert statements here... updating the test with size 2 assertion on actualFiles and between the (1) index item of actualFiles and expectedFiles :)
Thank you @szehon-ho !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123588098
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+
+ table.updateSchema().addColumn("category", Types.StringType.get()).commit();
+
+ List<Row> newRecords =
+ Lists.newArrayList(RowFactory.create(3, "b", "c"), RowFactory.create(4, "b", "c"));
+
+ StructType newSparkSchema =
+ SparkSchemaUtil.convert(
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "data", Types.StringType.get()),
+ optional(3, "category", Types.StringType.get())));
+
+ spark.createDataFrame(newRecords, newSparkSchema).coalesce(1).writeTo(tableName).append();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ Dataset<Row> actualFilesDs =
+ spark.sql(
+ "SELECT * FROM "
+ + tableName
+ + ".files VERSION AS OF "
+ + currentSnapshotId
+ + " ORDER BY content");
+ List<Row> actualFiles = TestHelpers.selectNonDerived(actualFilesDs).collectAsList();
+ Schema entriesTableSchema = Spark3Util.loadIcebergTable(spark, tableName + ".entries").schema();
+ List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
+ List<Record> expectedFiles =
+ expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
+
+ TestHelpers.assertEqualsSafe(
Review Comment:
Thanks, I'm really glad this worked.
Can we also add an assert on actualFiles size? And if its more than 0, could we add an assert for the other row too? (can do it in a for loop, i guess).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] syun64 commented on pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "syun64 (via GitHub)" <gi...@apache.org>.
syun64 commented on PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#issuecomment-1452654261
@szehon-ho I think that should cover all of the requested changes.. Please let me know if this is good to approve! Since this is a pretty important bug fix, I'm hoping we could slot it in for the next release...
I can create ones for Spark-3.2 and Spark-3.1 once you sign off on this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1123661886
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,60 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
Review Comment:
Throwing another one, testFilesTimeTravelAddedColumn , a bit shorter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] syun64 commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot ID
Posted by "syun64 (via GitHub)" <gi...@apache.org>.
syun64 commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1124976067
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +493,70 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesTableTimeTravelWithSchemaEvolution() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+
+ table.updateSchema().addColumn("category", Types.StringType.get()).commit();
+
+ List<Row> newRecords =
+ Lists.newArrayList(RowFactory.create(3, "b", "c"), RowFactory.create(4, "b", "c"));
+
+ StructType newSparkSchema =
+ SparkSchemaUtil.convert(
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "data", Types.StringType.get()),
+ optional(3, "category", Types.StringType.get())));
+
+ spark.createDataFrame(newRecords, newSparkSchema).coalesce(1).writeTo(tableName).append();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ Dataset<Row> actualFilesDs =
+ spark.sql(
+ "SELECT * FROM "
+ + tableName
+ + ".files VERSION AS OF "
+ + currentSnapshotId
+ + " ORDER BY content");
+ List<Row> actualFiles = TestHelpers.selectNonDerived(actualFilesDs).collectAsList();
+ Schema entriesTableSchema = Spark3Util.loadIcebergTable(spark, tableName + ".entries").schema();
+ List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
+ List<Record> expectedFiles =
+ expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
+
+ Assert.assertEquals("actualFiles size should be 1", 2, actualFiles.size());
Review Comment:
Appreciate all your help with these PRs @szehon-ho . Looking forward to 1.2.0 release and being able to work more with metadata tables :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6980: Spark-3.3: Bug Fix for Reading Metadata tables with Snapshot
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #6980:
URL: https://github.com/apache/iceberg/pull/6980#discussion_r1122487916
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMetadataTables.java:
##########
@@ -488,6 +489,54 @@ public void testMetadataLogEntries() throws Exception {
metadataLogWithProjection);
}
+ @Test
+ public void testFilesVersionAsOf() throws Exception {
+ // Create table and insert data
+ sql(
+ "CREATE TABLE %s (id bigint, data string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (data) "
+ + "TBLPROPERTIES"
+ + "('format-version'='2', 'write.delete.mode'='merge-on-read')",
+ tableName);
+
+ List<SimpleRecord> recordsA =
+ Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "a"));
+ spark
+ .createDataset(recordsA, Encoders.bean(SimpleRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+ Long olderSnapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s ADD COLUMNS (data2 string)",
+ tableName);
+
+ List<SimpleExtraColumnRecord> recordsB =
+ Lists.newArrayList(new SimpleExtraColumnRecord(1, "b", "c"), new SimpleExtraColumnRecord (2, "b", "c"));
+ spark
+ .createDataset(recordsB, Encoders.bean(SimpleExtraColumnRecord.class))
+ .coalesce(1)
+ .writeTo(tableName)
+ .append();
+
+
+ List<Object[]> res1 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, olderSnapshotId );
+
+ Dataset<Row> ds = spark.read().format("iceberg").option("snapshot-id", olderSnapshotId ).load(tableName + ".files");
+ List<Row> res2 = ds.collectAsList();
+
+ Long currentSnapshotId = table.currentSnapshot().snapshotId();
+
+ List<Object[]> res3 = sql("SELECT * from %s.files VERSION AS OF %s", tableName, currentSnapshotId);
+
+ Dataset<Row> ds2 = spark.read().format("iceberg").option("snapshot-id", currentSnapshotId ).load(tableName + ".files");
Review Comment:
We will actually need to do an assert here on the value, for a meaningful test, see the other examples.
You could look at other examples (testAllFilesPartitioned, which is more complete, but complex):
```
Dataset<Row> actualDataFilesDs =
spark.sql("SELECT * FROM " + tableName + ".files VERSION AS OF %s", tableName, currentSnasphotId);
List<ManifestFile> expectedDataManifests = TestHelpers.dataManifests(table);
List<Record> expectedDataFiles =
expectedEntries(table, FileContent.DATA, entriesTableSchema, expectedDataManifests, null);
TestHelpers.assertEqualsSafe(
SparkSchemaUtil.convert(TestHelpers.selectNonDerived(actualDataFilesDs).schema())
.asStruct(),
expectedDataFiles.get(0),
TestHelpers.selectNonDerived(actualDataFilesDs).collectAsList());
```
Or maybe easier, sort the query result and check fields individually , like testSnapshotReferencesMetatable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org