From b02d1e9a9d4c6abced97437a231f3262ec8edc6a Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 24 Jul 2026 12:48:02 -0400 Subject: [PATCH 1/9] Expand IcebergIO write API to cover all write options --- .../com/spotify/scio/iceberg/IcebergIO.scala | 19 +++++++++++---- .../syntax/IcebergSCollectionSyntax.scala | 23 ++++++++++++++----- .../spotify/scio/iceberg/IcebergIOTest.scala | 21 +++++++++++++++-- 3 files changed, 50 insertions(+), 13 deletions(-) diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala index daa616bfad..20bcba9e1c 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala @@ -41,6 +41,7 @@ private[scio] object ConfigMap { implicit val stringToMap: ConfigMapType[String] = _ => Map.empty implicit val intToMap: ConfigMapType[Int] = _ => Map.empty implicit val mapToMap: ConfigMapType[Map[String, String]] = _ => Map.empty + implicit val mapAnyRefToMap: ConfigMapType[Map[String, AnyRef]] = _ => Map.empty implicit val listToMap: ConfigMapType[List[String]] = _ => Map.empty implicit def optionToMap[T]: ConfigMapType[Option[T]] = _ => Map.empty @@ -86,8 +87,10 @@ final case class IcebergIO[T: RowType: Coder](table: String, catalogName: Option private[scio] def config(params: WriteP)(implicit mapper: ConfigMap.ConfigMapType[WriteP] - ): Map[String, AnyRef] = - baseConfig(params) + ): Map[String, AnyRef] = { + val extra = Option(params.extraConfigProperties).getOrElse(Map.empty) + baseConfig(params) - "extra_config_properties" ++ extra + } private[scio] def config( params: ReadP @@ -133,15 +136,21 @@ object IcebergIO { } case class WriteParam private ( catalogProperties: Map[String, String] = WriteParam.DefaultCatalogProperties, - configProperties: Map[String, String] = WriteParam.DefaultHadoopConfigProperties, + writeProperties: Map[String, String] = WriteParam.DefaultWriteProperties, + sortFields: List[String] = WriteParam.DefaultSortFields, + partitionFields: List[String] = WriteParam.DefaultPartitionFields, triggeringFrequencySeconds: Option[Int] = None, - directWriteByteLimit: Option[Int] = None + directWriteByteLimit: Option[Int] = None, + extraConfigProperties: Map[String, AnyRef] = WriteParam.DefaultExtraConfigProperties ) object WriteParam { val DefaultCatalogProperties: Map[String, String] = null - val DefaultHadoopConfigProperties: Map[String, String] = null + val DefaultWriteProperties: Map[String, String] = null + val DefaultSortFields: List[String] = null + val DefaultPartitionFields: List[String] = null val DefaultTriggeringFrequencySeconds: Int = -1 val DefaultDirectWriteByteLimit: Int = -1 + val DefaultExtraConfigProperties: Map[String, AnyRef] = null implicit val configMap: ConfigMap.ConfigMapType[WriteParam] = ConfigMap.gen[WriteParam] } diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala index 88d1d0efce..d192bee959 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala @@ -34,12 +34,18 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { * @param catalogProperties * any additional properties required by the Iceberg catalog; see: * https://iceberg.apache.org/docs/latest/catalog-properties - * @param hadoopConfigProperties - * any additional Hadoop configuration properties + * @param writeProperties + * any additional properties to pass to the Iceberg RecordWriter + * @param sortFields + * list of field names defining the sort order for written files + * @param partitionFields + * list of field names defining the partition spec for the table * @param triggeringFrequencySeconds * (streaming only) frequency at which snapshots are produced * @param directWriteByteLimit * (streaming only) limit for lifting bundles into the direct write path. + * @param extraConfigProperties + * additional properties to pass to the Managed IO config, i.e. `distribution_mode: hash` or `authosharding: true` * * For a complete reference, see: * https://docs.cloud.google.com/dataflow/docs/guides/managed-io-iceberg @@ -48,19 +54,24 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { table: String, catalogName: String = null, catalogProperties: Map[String, String] = IcebergIO.WriteParam.DefaultCatalogProperties, - hadoopConfigProperties: Map[String, String] = - IcebergIO.WriteParam.DefaultHadoopConfigProperties, + writeProperties: Map[String, String] = IcebergIO.WriteParam.DefaultWriteProperties, + sortFields: List[String] = IcebergIO.WriteParam.DefaultSortFields, + partitionFields: List[String] = IcebergIO.WriteParam.DefaultPartitionFields, + extraConfigProperties: Map[String, AnyRef] = IcebergIO.WriteParam.DefaultExtraConfigProperties, triggeringFrequencySeconds: Int = IcebergIO.WriteParam.DefaultTriggeringFrequencySeconds, directWriteByteLimit: Int = IcebergIO.WriteParam.DefaultDirectWriteByteLimit ): ClosedTap[Nothing] = { val params = IcebergIO.WriteParam( catalogProperties, - hadoopConfigProperties, + writeProperties, + sortFields, + partitionFields, Option(triggeringFrequencySeconds).filter( _ != IcebergIO.WriteParam.DefaultTriggeringFrequencySeconds ), - Option(directWriteByteLimit).filter(_ != IcebergIO.WriteParam.DefaultDirectWriteByteLimit) + Option(directWriteByteLimit).filter(_ != IcebergIO.WriteParam.DefaultDirectWriteByteLimit), + extraConfigProperties ) self.write(IcebergIO(table, Option(catalogName)))(params) } diff --git a/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala b/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala index 37f0305f12..af9b526c05 100644 --- a/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala +++ b/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala @@ -48,12 +48,16 @@ class IcebergIOTest extends ScioIOSpec { IcebergIO.WriteParam( Map.empty, Map.empty, + Nil, + Nil, None, None ), IcebergIO.WriteParam( Map("catalogProp1" -> "catalogProp1Value"), Map("configProp1" -> "configProp1Value", "configProp2" -> "configProp2Value"), + List("sortField1"), + List("partField1"), Some(10), Some(100) ) @@ -70,6 +74,9 @@ class IcebergIOTest extends ScioIOSpec { // reads "filter", // writes + "write_properties", + "sort_fields", + "partition_fields", "triggering_frequency_seconds", "direct_write_byte_limit" ) @@ -129,14 +136,24 @@ class IcebergIOTest extends ScioIOSpec { val writeParam = IcebergIO.WriteParam( Map("a" -> "b"), - Map("c" -> "d", "e" -> "f") + Map("c" -> "d", "e" -> "f"), + List("col1", "col2"), + List("partCol1"), + extraConfigProperties = Map( + "distribution_mode" -> "hash", + "autosharding" -> (true: java.lang.Boolean) + ) ) val managedConfig: Map[String, AnyRef] = io.config(writeParam) managedConfig should contain only ( - "config_properties" -> Map("c" -> "d", "e" -> "f"), + "write_properties" -> Map("c" -> "d", "e" -> "f"), + "sort_fields" -> List("col1", "col2"), + "partition_fields" -> List("partCol1"), "catalog_properties" -> Map("a" -> "b"), + "distribution_mode" -> "hash", + "autosharding" -> true, "table" -> "tableName", "catalog_name" -> "catalogName" ) From b47f6e5051a4b19b7e713934c0ecb65eb1294a7b Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 24 Jul 2026 12:48:56 -0400 Subject: [PATCH 2/9] scaladoc --- .../spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala index d192bee959..19bb454a8b 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala @@ -35,7 +35,8 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { * any additional properties required by the Iceberg catalog; see: * https://iceberg.apache.org/docs/latest/catalog-properties * @param writeProperties - * any additional properties to pass to the Iceberg RecordWriter + * any additional properties to pass to the Iceberg RecordWriter; see: + * https://iceberg.apache.org/docs/latest/configuration/#write-properties * @param sortFields * list of field names defining the sort order for written files * @param partitionFields From ca2dd9e1282c021413b230acf6d861d75fd8ed8c Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 24 Jul 2026 13:59:40 -0400 Subject: [PATCH 3/9] Integration test --- .../spotify/scio/iceberg/IcebergIOIT.scala | 120 +++++++++++++++++- 1 file changed, 116 insertions(+), 4 deletions(-) diff --git a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala index 0cb451ec71..c6c15364f5 100644 --- a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala +++ b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala @@ -17,6 +17,7 @@ package com.spotify.scio.iceberg import com.dimafeng.testcontainers.{ForAllTestContainer, GenericContainer} +import com.spotify.scio.parquet.BeamInputFile import com.spotify.scio.testing.PipelineSpec import magnolify.beam._ import org.apache.iceberg.catalog.{Namespace, TableIdentifier} @@ -28,7 +29,15 @@ import org.apache.iceberg.types.Types.{ StringType, StructType } -import org.apache.iceberg.{CatalogProperties, CatalogUtil, PartitionSpec, Schema} +import org.apache.iceberg.{ + CatalogProperties, + CatalogUtil, + NullOrder, + PartitionSpec, + Schema, + SortDirection +} +import org.apache.parquet.hadoop.ParquetFileReader import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy import java.time.Duration @@ -65,12 +74,15 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { lazy val uri = s"http://${container.containerIpAddress}:${container.mappedPort(ContainerPort)}" - override def afterStart(): Unit = { + lazy val catalog: RESTCatalog = { val cat = new RESTCatalog() cat.initialize(CatalogName, Map("uri" -> uri).asJava) + cat + } - cat.createNamespace(Namespace.of(NamespaceName)) - cat.createTable( + override def afterStart(): Unit = { + catalog.createNamespace(Namespace.of(NamespaceName)) + catalog.createTable( TableIdentifier.parse(TableName), new Schema( NestedField.required(0, "a", IntegerType.get()), @@ -104,4 +116,104 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { ) should containInAnyOrder(elements) } } + + it should "write with sort order" in { + val sortedTableName = s"${NamespaceName}.sorted_records" + val catalogProperties = Map( + CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, + CatalogProperties.URI -> uri + ) + val elements = 1.to(10).map(i => IcebergIOITRecord(i, s"$i", Nested(i % 2 == 0))) + + runWithRealContext() { sc => + sc.parallelize(elements) + .saveAsIceberg( + sortedTableName, + catalogProperties = catalogProperties, + sortFields = List("a asc nulls first") + ) + } + + val table = catalog.loadTable(TableIdentifier.parse(sortedTableName)) + val sortOrder = table.sortOrder() + sortOrder.isSorted shouldBe true + sortOrder.fields().size() shouldBe 1 + sortOrder.fields().get(0).direction() shouldBe SortDirection.ASC + sortOrder.fields().get(0).nullOrder() shouldBe NullOrder.NULLS_FIRST + } + + it should "write with partition spec" in { + val partitionedTableName = s"${NamespaceName}.partitioned_records" + val catalogProperties = Map( + CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, + CatalogProperties.URI -> uri + ) + val elements = 1.to(10).map(i => IcebergIOITRecord(i, s"$i", Nested(i % 2 == 0))) + + runWithRealContext() { sc => + sc.parallelize(elements) + .saveAsIceberg( + partitionedTableName, + catalogProperties = catalogProperties, + partitionFields = List("bucket(b, 2)") + ) + } + + val table = catalog.loadTable(TableIdentifier.parse(partitionedTableName)) + val spec = table.spec() + spec.isPartitioned shouldBe true + spec.fields().size() shouldBe 1 + spec.fields().get(0).name() shouldBe "b_bucket" + } + + it should "propagate Iceberg writeProperties" in { + val bfTableName = s"${NamespaceName}.bloom_filter_records" + val catalogProperties = Map( + CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, + CatalogProperties.URI -> uri + ) + val elements = 1.to(100).map(i => IcebergIOITRecord(i, s"value_$i", Nested(i % 2 == 0))) + + runWithRealContext() { sc => + sc.parallelize(elements) + .saveAsIceberg( + bfTableName, + catalogProperties = catalogProperties, + writeProperties = Map("write.parquet.bloom-filter-enabled.column.b" -> "true") + ) + } + + val table = catalog.loadTable(TableIdentifier.parse(bfTableName)) + val tasks = table.newScan().planFiles() + try { + val dataFiles = tasks.iterator().asScala.map(_.file().path().toString).toSeq + dataFiles should not be empty + + dataFiles.foreach { path => + val reader = ParquetFileReader.open(BeamInputFile.of(path)) + try { + reader.getFooter.getBlocks.asScala.foreach { block => + block.getColumns.asScala.foreach { col => + val hasBloom = col.getBloomFilterOffset > 0 + col.getPath.toDotString match { + case "b" => + // flip assertion once https://github.com/apache/beam/pull/39250/ is release in Beam 2.76 + withClue( + "Iceberg writeProperties are not supported in Beam versions <= 2.76. Once Beam is upgraded, flip this assertion to `true`." + ) { + hasBloom shouldBe false + } + case _ => + hasBloom shouldBe false + } + } + } + } finally { + reader.close() + } + } + } finally { + tasks.close() + } + } } From c1958b1aafa8de6f7e3b1ec1c53b41b3ecd2dcda Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 24 Jul 2026 14:48:49 -0400 Subject: [PATCH 4/9] fmt --- .../spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala index 19bb454a8b..dbe892bbe3 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala @@ -46,7 +46,8 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { * @param directWriteByteLimit * (streaming only) limit for lifting bundles into the direct write path. * @param extraConfigProperties - * additional properties to pass to the Managed IO config, i.e. `distribution_mode: hash` or `authosharding: true` + * additional properties to pass to the Managed IO config, i.e. `distribution_mode: hash` or + * `authosharding: true` * * For a complete reference, see: * https://docs.cloud.google.com/dataflow/docs/guides/managed-io-iceberg From 658d37d2b34d8e0137a146f8481f9fcc02586f5d Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Thu, 6 Aug 2026 15:21:13 -0400 Subject: [PATCH 5/9] Fix dynamic properties setter --- .../spotify/scio/iceberg/IcebergIOIT.scala | 73 ++++++------------- .../com/spotify/scio/iceberg/IcebergIO.scala | 4 +- .../syntax/IcebergSCollectionSyntax.scala | 10 +-- .../spotify/scio/iceberg/IcebergIOTest.scala | 4 +- 4 files changed, 30 insertions(+), 61 deletions(-) diff --git a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala index c6c15364f5..21dc7789bc 100644 --- a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala +++ b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala @@ -117,79 +117,53 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { } } - it should "write with sort order" in { - val sortedTableName = s"${NamespaceName}.sorted_records" + it should "propagate Iceberg dynamic table creation properties" in { + val tableName = s"${NamespaceName}.dynamic_table_creation" val catalogProperties = Map( CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, CatalogProperties.URI -> uri ) - val elements = 1.to(10).map(i => IcebergIOITRecord(i, s"$i", Nested(i % 2 == 0))) + val elements = 1.to(100).map(i => IcebergIOITRecord(i, s"value_$i", Nested(i % 2 == 0))) runWithRealContext() { sc => sc.parallelize(elements) .saveAsIceberg( - sortedTableName, + tableName, catalogProperties = catalogProperties, + tableProperties = Map( + "write.data.path" -> s"$tempDir/custom_path", + "write.parquet.bloom-filter-enabled.column.b" -> "true" + ), + partitionFields = List("bucket(b, 2)"), sortFields = List("a asc nulls first") ) } - val table = catalog.loadTable(TableIdentifier.parse(sortedTableName)) - val sortOrder = table.sortOrder() - sortOrder.isSorted shouldBe true - sortOrder.fields().size() shouldBe 1 - sortOrder.fields().get(0).direction() shouldBe SortDirection.ASC - sortOrder.fields().get(0).nullOrder() shouldBe NullOrder.NULLS_FIRST - } - - it should "write with partition spec" in { - val partitionedTableName = s"${NamespaceName}.partitioned_records" - val catalogProperties = Map( - CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, - CatalogProperties.URI -> uri - ) - val elements = 1.to(10).map(i => IcebergIOITRecord(i, s"$i", Nested(i % 2 == 0))) - - runWithRealContext() { sc => - sc.parallelize(elements) - .saveAsIceberg( - partitionedTableName, - catalogProperties = catalogProperties, - partitionFields = List("bucket(b, 2)") - ) - } + val table = catalog.loadTable(TableIdentifier.parse(tableName)) - val table = catalog.loadTable(TableIdentifier.parse(partitionedTableName)) + // Validate PartitionSpec val spec = table.spec() spec.isPartitioned shouldBe true spec.fields().size() shouldBe 1 spec.fields().get(0).name() shouldBe "b_bucket" - } - it should "propagate Iceberg writeProperties" in { - val bfTableName = s"${NamespaceName}.bloom_filter_records" - val catalogProperties = Map( - CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, - CatalogProperties.URI -> uri - ) - val elements = 1.to(100).map(i => IcebergIOITRecord(i, s"value_$i", Nested(i % 2 == 0))) + // Validate SortOrder + val sortOrder = table.sortOrder() + sortOrder.isSorted shouldBe true + sortOrder.fields().size() shouldBe 1 + sortOrder.fields().get(0).direction() shouldBe SortDirection.ASC + sortOrder.fields().get(0).nullOrder() shouldBe NullOrder.NULLS_FIRST - runWithRealContext() { sc => - sc.parallelize(elements) - .saveAsIceberg( - bfTableName, - catalogProperties = catalogProperties, - writeProperties = Map("write.parquet.bloom-filter-enabled.column.b" -> "true") - ) - } + // Validate table properties + table.properties().get("write.data.path") shouldBe s"$tempDir/custom_path" - val table = catalog.loadTable(TableIdentifier.parse(bfTableName)) val tasks = table.newScan().planFiles() try { val dataFiles = tasks.iterator().asScala.map(_.file().path().toString).toSeq dataFiles should not be empty dataFiles.foreach { path => + path should startWith(s"$tempDir/custom_path/") val reader = ParquetFileReader.open(BeamInputFile.of(path)) try { reader.getFooter.getBlocks.asScala.foreach { block => @@ -197,12 +171,7 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { val hasBloom = col.getBloomFilterOffset > 0 col.getPath.toDotString match { case "b" => - // flip assertion once https://github.com/apache/beam/pull/39250/ is release in Beam 2.76 - withClue( - "Iceberg writeProperties are not supported in Beam versions <= 2.76. Once Beam is upgraded, flip this assertion to `true`." - ) { - hasBloom shouldBe false - } + hasBloom shouldBe true case _ => hasBloom shouldBe false } diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala index 20bcba9e1c..bb1ec3ffe8 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/IcebergIO.scala @@ -136,7 +136,7 @@ object IcebergIO { } case class WriteParam private ( catalogProperties: Map[String, String] = WriteParam.DefaultCatalogProperties, - writeProperties: Map[String, String] = WriteParam.DefaultWriteProperties, + tableProperties: Map[String, String] = WriteParam.DefaultTableProperties, sortFields: List[String] = WriteParam.DefaultSortFields, partitionFields: List[String] = WriteParam.DefaultPartitionFields, triggeringFrequencySeconds: Option[Int] = None, @@ -145,7 +145,7 @@ object IcebergIO { ) object WriteParam { val DefaultCatalogProperties: Map[String, String] = null - val DefaultWriteProperties: Map[String, String] = null + val DefaultTableProperties: Map[String, String] = null val DefaultSortFields: List[String] = null val DefaultPartitionFields: List[String] = null val DefaultTriggeringFrequencySeconds: Int = -1 diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala index dbe892bbe3..475679164f 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala @@ -34,9 +34,9 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { * @param catalogProperties * any additional properties required by the Iceberg catalog; see: * https://iceberg.apache.org/docs/latest/catalog-properties - * @param writeProperties - * any additional properties to pass to the Iceberg RecordWriter; see: - * https://iceberg.apache.org/docs/latest/configuration/#write-properties + * @param tableProperties + * any additional Iceberg table properties to set during dynamic table creation; see: + * https://iceberg.apache.org/docs/latest/configuration/ * @param sortFields * list of field names defining the sort order for written files * @param partitionFields @@ -56,7 +56,7 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { table: String, catalogName: String = null, catalogProperties: Map[String, String] = IcebergIO.WriteParam.DefaultCatalogProperties, - writeProperties: Map[String, String] = IcebergIO.WriteParam.DefaultWriteProperties, + tableProperties: Map[String, String] = IcebergIO.WriteParam.DefaultTableProperties, sortFields: List[String] = IcebergIO.WriteParam.DefaultSortFields, partitionFields: List[String] = IcebergIO.WriteParam.DefaultPartitionFields, extraConfigProperties: Map[String, AnyRef] = IcebergIO.WriteParam.DefaultExtraConfigProperties, @@ -66,7 +66,7 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { val params = IcebergIO.WriteParam( catalogProperties, - writeProperties, + tableProperties, sortFields, partitionFields, Option(triggeringFrequencySeconds).filter( diff --git a/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala b/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala index af9b526c05..44bc1e0210 100644 --- a/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala +++ b/scio-managed/src/test/scala/com/spotify/scio/iceberg/IcebergIOTest.scala @@ -74,7 +74,7 @@ class IcebergIOTest extends ScioIOSpec { // reads "filter", // writes - "write_properties", + "table_properties", "sort_fields", "partition_fields", "triggering_frequency_seconds", @@ -148,7 +148,7 @@ class IcebergIOTest extends ScioIOSpec { val managedConfig: Map[String, AnyRef] = io.config(writeParam) managedConfig should contain only ( - "write_properties" -> Map("c" -> "d", "e" -> "f"), + "table_properties" -> Map("c" -> "d", "e" -> "f"), "sort_fields" -> List("col1", "col2"), "partition_fields" -> List("partCol1"), "catalog_properties" -> Map("a" -> "b"), From 92f2c2a401cd2eec28d4ff24dc0b6623341aad39 Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 7 Aug 2026 13:36:27 -0400 Subject: [PATCH 6/9] Clean up IT test --- .../spotify/scio/iceberg/IcebergIOIT.scala | 32 +++++++++---------- 1 file changed, 15 insertions(+), 17 deletions(-) diff --git a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala index 21dc7789bc..85b33a47bd 100644 --- a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala +++ b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala @@ -35,7 +35,7 @@ import org.apache.iceberg.{ NullOrder, PartitionSpec, Schema, - SortDirection + SortOrder } import org.apache.parquet.hadoop.ParquetFileReader import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy @@ -97,6 +97,8 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { ) } + override def beforeStop(): Unit = catalog.close() + "IcebergIO" should "work" in { val catalogProperties = Map( CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, @@ -125,13 +127,15 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { ) val elements = 1.to(100).map(i => IcebergIOITRecord(i, s"value_$i", Nested(i % 2 == 0))) + val customWriteDataPath = s"$tempDir/custom_path" + runWithRealContext() { sc => sc.parallelize(elements) .saveAsIceberg( tableName, catalogProperties = catalogProperties, tableProperties = Map( - "write.data.path" -> s"$tempDir/custom_path", + "write.data.path" -> customWriteDataPath, "write.parquet.bloom-filter-enabled.column.b" -> "true" ), partitionFields = List("bucket(b, 2)"), @@ -141,29 +145,23 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { val table = catalog.loadTable(TableIdentifier.parse(tableName)) - // Validate PartitionSpec - val spec = table.spec() - spec.isPartitioned shouldBe true - spec.fields().size() shouldBe 1 - spec.fields().get(0).name() shouldBe "b_bucket" - - // Validate SortOrder - val sortOrder = table.sortOrder() - sortOrder.isSorted shouldBe true - sortOrder.fields().size() shouldBe 1 - sortOrder.fields().get(0).direction() shouldBe SortDirection.ASC - sortOrder.fields().get(0).nullOrder() shouldBe NullOrder.NULLS_FIRST + // Validate PartitionSpec and SortOrder + table.spec() shouldEqual PartitionSpec.builderFor(table.schema()).bucket("b", 2).build() + table.sortOrder() shouldEqual SortOrder + .builderFor(table.schema()) + .asc("a", NullOrder.NULLS_FIRST) + .build() // Validate table properties - table.properties().get("write.data.path") shouldBe s"$tempDir/custom_path" + table.properties().get("write.data.path") shouldBe customWriteDataPath val tasks = table.newScan().planFiles() try { - val dataFiles = tasks.iterator().asScala.map(_.file().path().toString).toSeq + val dataFiles = tasks.iterator().asScala.map(_.file().location()).toSeq dataFiles should not be empty dataFiles.foreach { path => - path should startWith(s"$tempDir/custom_path/") + path should startWith(s"$customWriteDataPath/b_bucket=") val reader = ParquetFileReader.open(BeamInputFile.of(path)) try { reader.getFooter.getBlocks.asScala.foreach { block => From 445269ba616487ad74aae08b9adcb4366e36e060 Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 7 Aug 2026 13:41:42 -0400 Subject: [PATCH 7/9] Update doc --- .../spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala index 475679164f..3155433b55 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala @@ -36,7 +36,7 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { * https://iceberg.apache.org/docs/latest/catalog-properties * @param tableProperties * any additional Iceberg table properties to set during dynamic table creation; see: - * https://iceberg.apache.org/docs/latest/configuration/ + * https://iceberg.apache.org/docs/latest/configuration/#write-properties * @param sortFields * list of field names defining the sort order for written files * @param partitionFields From e3a718ac80facbdea00c727f095063212894bef2 Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 7 Aug 2026 14:02:55 -0400 Subject: [PATCH 8/9] Typo in doc --- .../spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala index 3155433b55..2add91d0c7 100644 --- a/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala +++ b/scio-managed/src/main/scala/com/spotify/scio/iceberg/syntax/IcebergSCollectionSyntax.scala @@ -47,7 +47,7 @@ class IcebergSCollectionSyntax[T: RowType: Coder](self: SCollection[T]) { * (streaming only) limit for lifting bundles into the direct write path. * @param extraConfigProperties * additional properties to pass to the Managed IO config, i.e. `distribution_mode: hash` or - * `authosharding: true` + * `autosharding: true` * * For a complete reference, see: * https://docs.cloud.google.com/dataflow/docs/guides/managed-io-iceberg From 4e2686c851856a25d0d18b4fdfbb29d9113b2b6f Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Fri, 7 Aug 2026 14:12:34 -0400 Subject: [PATCH 9/9] schema should be 1-indexed --- .../spotify/scio/iceberg/IcebergIOIT.scala | 43 +++++++++++-------- 1 file changed, 26 insertions(+), 17 deletions(-) diff --git a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala index 85b33a47bd..fa6d33a01a 100644 --- a/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala +++ b/integration/src/test/scala/com/spotify/scio/iceberg/IcebergIOIT.scala @@ -20,6 +20,7 @@ import com.dimafeng.testcontainers.{ForAllTestContainer, GenericContainer} import com.spotify.scio.parquet.BeamInputFile import com.spotify.scio.testing.PipelineSpec import magnolify.beam._ +import magnolify.beam.logical.millis._ import org.apache.iceberg.catalog.{Namespace, TableIdentifier} import org.apache.iceberg.rest.RESTCatalog import org.apache.iceberg.types.Types.{ @@ -27,7 +28,8 @@ import org.apache.iceberg.types.Types.{ IntegerType, NestedField, StringType, - StructType + StructType, + TimestampType } import org.apache.iceberg.{ CatalogProperties, @@ -40,13 +42,14 @@ import org.apache.iceberg.{ import org.apache.parquet.hadoop.ParquetFileReader import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy -import java.time.Duration +import java.time.{Duration, Instant} import java.io.File import java.nio.file.Files +import java.time.temporal.ChronoUnit import scala.jdk.CollectionConverters._ case class Nested(d: Boolean) -case class IcebergIOITRecord(a: Int, b: String, c: Nested) +case class IcebergIOITRecord(ts: Instant, a: Int, b: String, c: Nested) object IcebergIOITRecord { implicit val icebergIOITRecordRowType: RowType[IcebergIOITRecord] = RowType[IcebergIOITRecord] } @@ -74,6 +77,17 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { lazy val uri = s"http://${container.containerIpAddress}:${container.mappedPort(ContainerPort)}" + lazy val tableSchema = new Schema( + NestedField.required(1, "ts", TimestampType.withZone()), + NestedField.required(2, "a", IntegerType.get()), + NestedField.required(3, "b", StringType.get()), + NestedField.required( + 4, + "c", + StructType.of(NestedField.required(5, "d", BooleanType.get())) + ) + ) + lazy val catalog: RESTCatalog = { val cat = new RESTCatalog() cat.initialize(CatalogName, Map("uri" -> uri).asJava) @@ -84,15 +98,7 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { catalog.createNamespace(Namespace.of(NamespaceName)) catalog.createTable( TableIdentifier.parse(TableName), - new Schema( - NestedField.required(0, "a", IntegerType.get()), - NestedField.required(1, "b", StringType.get()), - NestedField.required( - 2, - "c", - StructType.of(NestedField.required(3, "d", BooleanType.get())) - ) - ), + tableSchema, PartitionSpec.unpartitioned() ) } @@ -104,7 +110,8 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, CatalogProperties.URI -> uri ) - val elements = 1.to(10).map(i => IcebergIOITRecord(i, s"$i", Nested(i % 2 == 0))) + val ts = Instant.now().truncatedTo(ChronoUnit.DAYS) + val elements = 1.to(10).map(i => IcebergIOITRecord(ts, i, s"$i", Nested(i % 2 == 0))) runWithRealContext() { sc => sc.parallelize(elements) @@ -125,7 +132,8 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { CatalogUtil.ICEBERG_CATALOG_TYPE -> CatalogUtil.ICEBERG_CATALOG_TYPE_REST, CatalogProperties.URI -> uri ) - val elements = 1.to(100).map(i => IcebergIOITRecord(i, s"value_$i", Nested(i % 2 == 0))) + val elements = + 1.to(100).map(i => IcebergIOITRecord(Instant.now(), i, s"value_$i", Nested(i % 2 == 0))) val customWriteDataPath = s"$tempDir/custom_path" @@ -138,15 +146,16 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { "write.data.path" -> customWriteDataPath, "write.parquet.bloom-filter-enabled.column.b" -> "true" ), - partitionFields = List("bucket(b, 2)"), + partitionFields = List("day(ts)"), sortFields = List("a asc nulls first") ) } val table = catalog.loadTable(TableIdentifier.parse(tableName)) + table.schema().sameSchema(tableSchema) shouldBe true // Validate PartitionSpec and SortOrder - table.spec() shouldEqual PartitionSpec.builderFor(table.schema()).bucket("b", 2).build() + table.spec() shouldEqual PartitionSpec.builderFor(table.schema()).day("ts").build() table.sortOrder() shouldEqual SortOrder .builderFor(table.schema()) .asc("a", NullOrder.NULLS_FIRST) @@ -161,7 +170,7 @@ class IcebergIOIT extends PipelineSpec with ForAllTestContainer { dataFiles should not be empty dataFiles.foreach { path => - path should startWith(s"$customWriteDataPath/b_bucket=") + path should startWith(s"$customWriteDataPath/ts_day=") val reader = ParquetFileReader.open(BeamInputFile.of(path)) try { reader.getFooter.getBlocks.asScala.foreach { block =>