hsiang-c commented on code in PR #4752:
URL: https://github.com/apache/datafusion-comet/pull/4752#discussion_r3553459114
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -4310,32 +4310,235 @@ class CometIcebergNativeSuite
}
}
- test("CometScanRule should report unsupported metadata columns") {
+ test("metadata columns - _pos returns row position within each file") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
withTempIcebergDir { warehouseDir =>
withSQLConf(
"spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
"spark.sql.catalog.test_cat.type" -> "hadoop",
"spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
CometConf.COMET_ENABLED.key -> "true",
CometConf.COMET_EXEC_ENABLED.key -> "true",
- CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true",
- CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true") {
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.pos_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, name STRING) USING iceberg
+ """)
+
+ spark.sql(s"""
+ INSERT INTO $table VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Charlie')
+ """)
+
+ checkIcebergNativeScan(s"SELECT id, _pos FROM $table ORDER BY id")
+
+ // _pos should be 0-indexed per file
+ val result = spark.sql(s"SELECT id, _pos FROM $table ORDER BY
id").collect()
+ assert(result(0).getLong(1) == 0L)
+ assert(result(1).getLong(1) == 1L)
+ assert(result(2).getLong(1) == 2L)
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _pos resets per file with multiple data files") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
- val table = "test_cat.db.test_meta_cols"
+ val table = "test_cat.db.pos_multi_file_test"
try {
spark.sql(s"""
CREATE TABLE $table (id INT, value DOUBLE) USING iceberg
+ """)
+
+ // Two separate inserts create two data files
+ spark.sql(s"INSERT INTO $table VALUES (1, 10.0), (2, 20.0), (3,
30.0)")
+ spark.sql(s"INSERT INTO $table VALUES (4, 40.0), (5, 50.0)")
+
+ // _pos resets to 0 at the start of each file
+ checkIcebergNativeScan(s"SELECT id, _pos, _file FROM $table ORDER BY
_file, _pos")
+
+ // Verify _pos starts at 0 in each file
+ val result = spark.sql(s"SELECT _pos FROM $table ORDER BY _file,
_pos").collect()
+ val positions = result.map(_.getLong(0))
+ // First file has 0,1,2; second file has 0,1 (or vice versa)
+ assert(positions.contains(0L))
+ assert(positions.count(_ == 0L) == 2, "Each file should start _pos
at 0")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _spec_id returns partition spec ID") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.spec_id_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, category STRING, value DOUBLE)
+ USING iceberg PARTITIONED BY (category)
+ """)
+
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (1, 'A', 10.0), (2, 'B', 20.0), (3, 'A', 30.0)
+ """)
+
+ checkIcebergNativeScan(s"SELECT id, _spec_id FROM $table ORDER BY
id")
+
+ // All rows written under the initial spec should have _spec_id = 0
+ val result = spark.sql(s"SELECT DISTINCT _spec_id FROM
$table").collect()
+ assert(result.length == 1)
+ assert(result(0).getInt(0) == 0)
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _spec_id changes after partition evolution") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+
+ val table = "test_cat.db.spec_id_evolution"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, region STRING, category STRING)
+ USING iceberg PARTITIONED BY (region)
+ """)
+
+ // Data under spec 0
+ spark.sql(s"INSERT INTO $table VALUES (1, 'US', 'A'), (2, 'EU',
'B')")
+
+ // Evolve partition spec: add category field -> spec 1
Review Comment:
(nit) We might be able to use SparkSQL as well
```scala
ALTER TABLE $table ADD PARTITION FIELD category;
```
--
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]
