yabola commented on code in PR #6962:
URL: https://github.com/apache/iceberg/pull/6962#discussion_r1161231991
##########
spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java:
##########
@@ -1765,6 +1767,67 @@ public void testAllManifestTableSnapshotFiltering()
throws Exception {
}
}
+ @Test
+ public void testTableWithInt96Timestamp() throws IOException {
+ File parquetTableDir = temp.newFolder("table_timestamp_int96");
+ String parquetTableLocation = parquetTableDir.toURI().toString();
+ Schema schema =
+ new Schema(
+ optional(1, "id", Types.LongType.get()),
+ optional(2, "tmp_col", Types.TimestampType.withZone()));
+ spark.conf().set(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE().key(), "INT96");
+
+ LocalDateTime start = LocalDateTime.of(2000, 1, 31, 0, 0, 0);
+ LocalDateTime end = LocalDateTime.of(2100, 1, 1, 0, 0, 0);
+ long startSec = start.toEpochSecond(ZoneOffset.UTC);
+ long endSec = end.toEpochSecond(ZoneOffset.UTC);
+ Column idColumn = functions.expr("id");
+ Column secondsColumn =
+ functions.expr("(id % " + (endSec - startSec) + " + " + startSec +
")").as("seconds");
+ Column timestampColumn = functions.expr("cast( seconds as timestamp) as
tmp_col");
+
+ for (Boolean useDict : new Boolean[] {true, false}) {
+ for (Boolean useVectorization : new Boolean[] {true, false}) {
+ spark.sql("DROP TABLE IF EXISTS parquet_table");
+ spark
+ .range(0, 5000, 100, 1)
+ .select(idColumn, secondsColumn)
+ .select(idColumn, timestampColumn)
+ .write()
+ .format("parquet")
+ .option("parquet.enable.dictionary", useDict)
+ .mode("overwrite")
+ .option("path", parquetTableLocation)
+ .saveAsTable("parquet_table");
+ TableIdentifier tableIdentifier = TableIdentifier.of("db",
"table_with_timestamp_int96");
+ Table table = createTable(tableIdentifier, schema,
PartitionSpec.unpartitioned());
+ table
+ .updateProperties()
+ .set(TableProperties.PARQUET_VECTORIZATION_ENABLED,
useVectorization.toString())
+ .commit();
+
+ String stagingLocation = table.location() + "/metadata";
+ SparkTableUtil.importSparkTable(
+ spark,
+ new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"),
+ table,
+ stagingLocation);
+
+ // validate we get the expected results back
+ List<Row> expected =
spark.table("parquet_table").select("tmp_col").collectAsList();
+ List<Row> actual =
+ spark
+ .read()
+ .format("iceberg")
+ .load(loadLocation(tableIdentifier))
+ .select("tmp_col")
+ .collectAsList();
+ Assert.assertEquals("Rows must match", expected, actual);
Review Comment:
done. Thank you very much for your testing on Delta !
##########
spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java:
##########
@@ -1765,6 +1767,67 @@ public void testAllManifestTableSnapshotFiltering()
throws Exception {
}
}
+ @Test
+ public void testTableWithInt96Timestamp() throws IOException {
+ File parquetTableDir = temp.newFolder("table_timestamp_int96");
+ String parquetTableLocation = parquetTableDir.toURI().toString();
+ Schema schema =
+ new Schema(
+ optional(1, "id", Types.LongType.get()),
+ optional(2, "tmp_col", Types.TimestampType.withZone()));
+ spark.conf().set(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE().key(), "INT96");
+
+ LocalDateTime start = LocalDateTime.of(2000, 1, 31, 0, 0, 0);
+ LocalDateTime end = LocalDateTime.of(2100, 1, 1, 0, 0, 0);
+ long startSec = start.toEpochSecond(ZoneOffset.UTC);
+ long endSec = end.toEpochSecond(ZoneOffset.UTC);
+ Column idColumn = functions.expr("id");
+ Column secondsColumn =
+ functions.expr("(id % " + (endSec - startSec) + " + " + startSec +
")").as("seconds");
+ Column timestampColumn = functions.expr("cast( seconds as timestamp) as
tmp_col");
+
+ for (Boolean useDict : new Boolean[] {true, false}) {
+ for (Boolean useVectorization : new Boolean[] {true, false}) {
+ spark.sql("DROP TABLE IF EXISTS parquet_table");
+ spark
+ .range(0, 5000, 100, 1)
+ .select(idColumn, secondsColumn)
+ .select(idColumn, timestampColumn)
+ .write()
+ .format("parquet")
+ .option("parquet.enable.dictionary", useDict)
+ .mode("overwrite")
+ .option("path", parquetTableLocation)
+ .saveAsTable("parquet_table");
+ TableIdentifier tableIdentifier = TableIdentifier.of("db",
"table_with_timestamp_int96");
+ Table table = createTable(tableIdentifier, schema,
PartitionSpec.unpartitioned());
+ table
+ .updateProperties()
+ .set(TableProperties.PARQUET_VECTORIZATION_ENABLED,
useVectorization.toString())
+ .commit();
+
+ String stagingLocation = table.location() + "/metadata";
+ SparkTableUtil.importSparkTable(
+ spark,
+ new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"),
+ table,
+ stagingLocation);
+
+ // validate we get the expected results back
+ List<Row> expected =
spark.table("parquet_table").select("tmp_col").collectAsList();
+ List<Row> actual =
+ spark
+ .read()
+ .format("iceberg")
+ .load(loadLocation(tableIdentifier))
+ .select("tmp_col")
+ .collectAsList();
+ Assert.assertEquals("Rows must match", expected, actual);
Review Comment:
done. Thank you very much for testing on Delta !
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
