mbutrovich commented on code in PR #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331#discussion_r3830848772
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -86,6 +86,24 @@ pub struct IcebergScanExec {
tasks: Vec<FileScanTask>,
/// Number of data files to read concurrently
data_file_concurrency_limit: usize,
+ /// FileIO (and, for S3, the JVM credential bridge behind it) built once
at plan time and shared
+ /// across partitions. FileIO is cheap to clone (Arc-backed), so each
`execute` clones this
+ /// rather than rebuilding the storage factory + credential bridge. This
matters in the ordered
+ /// path, where the scan is one partition per file and `execute` is called
once per file.
+ file_io: FileIO,
+ /// Table sort order Iceberg reported, translated against `output_schema`.
`Some` makes this a
+ /// multi-partition scan: one sorted stream per task, which a
SortPreservingMergeExec above
+ /// merges back into one sorted partition. It is also advertised in
`plan_properties`. `None`
+ /// keeps the old single-partition unordered read (all tasks streamed
together).
+ ///
+ /// Concurrency note: in the ordered path each partition reads exactly one
task, so
+ /// `data_file_concurrency_limit` no longer bounds cross-file concurrency;
instead the wrapping
+ /// SortPreservingMergeExec drives one reader per file to merge them. That
fan-out (files per
+ /// Spark partition) is intrinsic to a k-way merge of per-file sorted
streams -- the files must
Review Comment:
Following up on the fan-out discussion: I don't think this needs a wholly
new algorithm, but it does need more than `IcebergScanExec` alone, since the
operator that decides when to poll each partition is `SortPreservingMergeExec`,
not this one. That operator has no notion of a bound and primes every input
stream up front because polling is the only signal it has for "what's in this
stream." Making `IcebergScanExec::execute` lazy internally doesn't avoid that,
since `SortPreservingMergeExec` still calls `poll_next` on every partition
almost immediately to seed its loser tree.
What would actually bound this is a custom merge operator that takes each
file's min/max bound on the sort key alongside its `FileScanTask`, keeps a
small active set merged the way `SortPreservingMergeExec` does today, and holds
a priority queue of unopened files ordered by min bound, only calling
`execute`/`poll_next` on the next pending file once its min bound could produce
the next output value. That bounds concurrently-open readers by the overlap
width of the file ranges at a given point in the merge rather than by total
file count, which is the number that actually blows up on a sorted table with a
lot of small commits.
The blocker today is data, not algorithm: `FileScanTask` in iceberg-rust
doesn't carry column-level bounds. They exist upstream on the manifest-entry
`DataFile` (`lower_bounds`/`upper_bounds`, already used for predicate
pushdown), just not threaded down into the task struct the native scan gets. So
this needs plumbing in iceberg-rust or an extra field fetched at scan-planning
time on the JVM side, either way before a bound-driven admission scheme can
work.
Given the risk, could we land @andygrove's simpler mitigation in this PR, a
config for the max files to merge per partition that falls back to the
unordered read above it, and keep the bound-driven design above as the concrete
plan for #5343? Sort-merge ships on by default, and the workload it targets (a
sorted table with accumulated small commits) is exactly where a partition ends
up with hundreds of files, so I'd rather the default be safe now and the deeper
optimization follow once the stats plumbing exists.
##########
spark/src/test/scala/org/apache/comet/CometIcebergSortMergeReadSuite.scala:
##########
@@ -0,0 +1,641 @@
+/*
+ * 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.comet
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.catalyst.expressions.{Add, Ascending,
AttributeReference, Literal, SortOrder}
+import org.apache.spark.sql.comet.{CometIcebergNativeScanExec, CometSortExec}
+import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
+import org.apache.spark.sql.execution.{SortExec, SparkPlan}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.types.IntegerType
+
+import org.apache.comet.serde.operator.CometIcebergNativeScan
+
+/**
+ * Tests for the sort-aware native Iceberg scan (branch `stream-merge`): the
scan reports the
+ * Iceberg table sort order to Spark and does a per-partition streaming k-way
merge of the
+ * already-sorted files, so Catalyst can drop the Sort and (with
storage-partitioned join) the
+ * Exchange that a sort-merge join / grouped aggregate / window would
otherwise need.
+ *
+ * The suite has two parts:
+ * - Unit tests of the [[CometIcebergNativeScan.reportableOrdering]] gate
(no SparkSession
+ * required) -- the single decision shared by the proto serialization
(which turns on the
+ * native SortPreservingMergeExec) and
CometIcebergNativeScanExec.outputOrdering (which tells
+ * Spark the scan is sorted).
+ * - End-to-end tests over real Iceberg tables that exercise every Spark 4.0
mechanism which
+ * exploits already-sorted input to avoid a Sort or a shuffle (all keyed
off
+ * `SortOrder.orderingSatisfies`, prefix semantics):
`SupportsReportOrdering` ->
+ * `BatchScanExec.outputOrdering` and `EnsureRequirements` eliding the
required-child Sort;
+ * `EliminateSorts` / `RemoveRedundantSorts`;
`SortMergeJoinExec.requiredChildOrdering` +
+ * storage-partitioned join (`KeyGroupedPartitioning`,
+ * `spark.sql.sources.v2.bucketing.enabled`); `ReplaceHashWithSortAgg` /
`SortAggregateExec`;
+ * `WindowExec`; `TakeOrderedAndProjectExec`.
+ *
+ * Two invariants determine the end-to-end assertions:
+ * 1. Correctness is checked unconditionally via `checkSparkAnswer` (Comet
vs vanilla Spark).
+ * This is the primary guarantee: any k-way-merge defect (dropped,
duplicated or mis-ordered
+ * rows, or an outputOrdering/outputPartitioning that does not match the
rows the native
+ * operator actually produces) shows up as a result mismatch. It holds on
any Iceberg build.
+ * 2. The strict "no Sort / no Exchange" plan assertions are the *target*
of this feature.
+ * They only hold where the Iceberg build actually reports the ordering
+ * (`SupportsReportOrdering`, today an Iceberg fork feature -- the
published/upstream Iceberg
+ * used in CI does not report it) and where the native scan reports
`KeyGroupedPartitioning`.
+ * Each such test therefore runs the correctness check first, then
`assume`s the reporting is
+ * active before asserting the plan shape, so it enforces the contract on
a reporting build
+ * and is skipped (not failed) elsewhere. `sort = 0` is asserted only for
*operator-required*
+ * orderings (SMJ / aggregate / window), never for a global `ORDER BY`,
which Spark keeps
+ * regardless (the per-partition merge is not a global order).
+ *
+ * These cannot be SQL-file fixtures: setting an Iceberg sort order needs the
Iceberg Java API
+ * (the Comet test session registers no Iceberg SQL extensions, so `WRITE
ORDERED BY` will not
+ * parse), and the plan-shape assertions need access to the executed SparkPlan.
+ *
+ * Each test gets its own catalog name and temp warehouse (the Hadoop
`SparkCatalog` instance is
+ * cached per catalog name, so a shared name would bind every test to the
first warehouse), and
+ * its tables are dropped in a `finally` so a failing test cannot leak a table
into a later one.
+ */
+class CometIcebergSortMergeReadSuite
+ extends CometTestBase
+ with CometIcebergTestBase
+ with AdaptiveSparkPlanHelper {
+
+ //
---------------------------------------------------------------------------------------------
+ // Unit tests of the reportableOrdering gate (merged from the former
CometIcebergSortMergeSuite).
+ // v1 identity-scope gate as a pure function, independent of a live
SparkSession or an Iceberg
+ // build that reports ordering. The flag defaults to enabled, so no SQLConf
override is needed.
+ //
---------------------------------------------------------------------------------------------
+
+ private val gateA = AttributeReference("a", IntegerType)()
+ private val gateB = AttributeReference("b", IntegerType)()
+
+ test("gate: identity ordering on projected columns is reportable") {
+ val ordering = Seq(SortOrder(gateA, Ascending))
+ assert(
+ CometIcebergNativeScan.reportableOrdering(Some(ordering), Seq(gateA,
gateB)) === ordering)
+ }
+
+ test("gate: ordering on a column outside the projection falls back") {
+ val ordering = Seq(SortOrder(gateA, Ascending))
+ assert(CometIcebergNativeScan.reportableOrdering(Some(ordering),
Seq(gateB)).isEmpty)
+ }
+
+ test("gate: a transform (non-AttributeReference) sort child falls back") {
+ val ordering = Seq(SortOrder(Add(gateA, Literal(1)), Ascending))
+ assert(CometIcebergNativeScan.reportableOrdering(Some(ordering),
Seq(gateA)).isEmpty)
+ }
+
+ test("gate: if any sort field is unreportable, the whole ordering falls
back") {
+ val ordering = Seq(SortOrder(gateA, Ascending), SortOrder(Add(gateB,
Literal(1)), Ascending))
+ assert(CometIcebergNativeScan.reportableOrdering(Some(ordering),
Seq(gateA, gateB)).isEmpty)
+ }
+
+ test("gate: absent or empty ordering falls back") {
+ assert(CometIcebergNativeScan.reportableOrdering(None, Seq(gateA)).isEmpty)
+ assert(CometIcebergNativeScan.reportableOrdering(Some(Seq.empty),
Seq(gateA)).isEmpty)
+ }
+
+ test("gate: a sort key on an ordering-unsafe column (e.g. UUID) falls back")
{
+ // "a" stands in for a UUID column: Iceberg maps it to StringType but
sorts by its own order,
+ // so it is unsafe to honour even though it looks like a plain string at
the Spark level.
+ val ordering = Seq(SortOrder(gateA, Ascending))
+ assert(
+ CometIcebergNativeScan
+ .reportableOrdering(Some(ordering), Seq(gateA, gateB), Set("a"))
+ .isEmpty)
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // End-to-end fixtures.
+ //
---------------------------------------------------------------------------------------------
+
+ private val catalogCounter = new AtomicInteger(0)
+
+ // preserve-data-ordering makes Iceberg report the table sort order;
adaptive off keeps the
+ // executed plan stable for the sort/shuffle counts below.
+ private val orderedReadConf: Seq[(String, String)] = Seq(
+ "spark.sql.iceberg.planning.preserve-data-ordering" -> "true",
+ "spark.sql.adaptive.enabled" -> "false")
+
+ // Storage-partitioned join config. preserve-data-grouping + v2 bucketing
let Iceberg report
+ // KeyGroupedPartitioning on the BatchScanExec, and the join knobs force a
sort-merge join over
+ // co-partitioned inputs so the Exchange can be eliminated. Comet does not
report partitioning
+ // itself; Spark's EnsureRequirements eliminates the shuffle on the
BatchScanExec before Comet
+ // converts the scan. Adaptive off keeps the executed plan stable for the
counts below.
+ private val spjConf: Seq[(String, String)] = Seq(
+ "spark.sql.iceberg.planning.preserve-data-ordering" -> "true",
+ "spark.sql.iceberg.planning.preserve-data-grouping" -> "true",
+ "spark.sql.sources.v2.bucketing.enabled" -> "true",
+ "spark.sql.sources.v2.bucketing.pushPartValues.enabled" -> "true",
+ "spark.sql.requireAllClusterKeysForCoPartition" -> "false",
+ "spark.sql.autoBroadcastJoinThreshold" -> "-1",
+ "spark.sql.join.preferSortMergeJoin" -> "true",
+ "spark.sql.adaptive.enabled" -> "false")
+
+ /**
+ * Runs `f` against a fresh, uniquely-named Hadoop catalog backed by a fresh
temp warehouse,
+ * then drops the named tables (IF EXISTS) in a `finally` -- so tables are
cleaned up even when
+ * the test body fails, and no two tests can collide on a table name. `f`
receives the catalog
+ * name; tables live under the `db` namespace, e.g. `$cat.db.$table`.
+ */
+ private def withSortedTables(extraConf: Seq[(String, String)])(tables:
String*)(
+ f: String => Unit): Unit = {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ val cat = s"sort_cat_${catalogCounter.incrementAndGet()}"
+ val cometConf = Seq(
+ s"spark.sql.catalog.$cat" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$cat.type" -> "hadoop",
+ s"spark.sql.catalog.$cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")
+ withSQLConf((cometConf ++ extraConf): _*) {
+ try f(cat)
+ finally tables.foreach(t => spark.sql(s"DROP TABLE IF EXISTS
$cat.db.$t"))
+ }
+ }
+ }
+
+ /**
+ * Sets the table sort order via the Iceberg Java API. `cols` is (column,
ascending); ascending
+ * uses Iceberg's default NULLS FIRST and descending its default NULLS LAST,
matching the ORDER
+ * BY null-ordering the tests use.
+ */
+ private def replaceSortOrder(
+ cat: String,
+ namespace: String,
+ table: String,
+ cols: (String, Boolean)*): Unit = {
+ val catalog = spark.sessionState.catalogManager
+ .catalog(cat)
+ .asInstanceOf[org.apache.iceberg.spark.SparkCatalog]
+ val ident =
+ org.apache.spark.sql.connector.catalog.Identifier.of(Array(namespace),
table)
+ val icebergTable = catalog
+ .loadTable(ident)
+ .asInstanceOf[org.apache.iceberg.spark.source.SparkTable]
+ .table()
+ var sortOrder = icebergTable.replaceSortOrder()
+ cols.foreach { case (c, asc) =>
+ sortOrder = if (asc) sortOrder.asc(c) else sortOrder.desc(c)
+ }
+ sortOrder.commit()
+ }
+
+ /**
+ * Each string becomes a separate INSERT, hence a separate data file, so
merging is required.
+ */
+ private def insertBatches(cat: String, table: String, batches: String*):
Unit =
+ batches.foreach(values => spark.sql(s"INSERT INTO $cat.db.$table VALUES
$values"))
+
+ private def nativeScans(plan: SparkPlan): Seq[CometIcebergNativeScanExec] =
+ collect(stripAQEPlan(plan)) { case s: CometIcebergNativeScanExec => s }
+
+ private def countSorts(plan: SparkPlan): Int =
+ collect(stripAQEPlan(plan)) {
+ case s: SortExec => s
+ case s: CometSortExec => s
+ }.size
+
+ private def countShuffles(plan: SparkPlan): Int =
+ collect(stripAQEPlan(plan)) {
+ case e: ShuffleExchangeExec => e
+ case e: CometShuffleExchangeExec => e
+ }.size
+
+ /** True once every native scan in the plan advertises the reported
ordering. */
+ private def orderingReported(plan: SparkPlan): Boolean = {
+ val scans = nativeScans(plan)
+ scans.nonEmpty && scans.forall(_.outputOrdering.nonEmpty)
+ }
+
+ // NOTE ON CANCELED TESTS: the helper below CANCELS the test (ScalaTest
`assume`, reported as
+ // "!!! CANCELED !!!", not a failure) when the Iceberg build on the
classpath does not implement
+ // the DSv2 `SupportsReportOrdering` API. Published/upstream Iceberg (the
runtime used in CI and
+ // the default mvn profiles) does not report a sort order, so
`outputOrdering` comes back empty
+ // and there is no eliminated Sort to assert on. These tests therefore show
as canceled there --
+ // that is expected, NOT a regression. The preceding `checkSparkAnswer` has
already validated
+ // correctness; only the sort/shuffle-elimination plan assertion is skipped.
Run against an
+ // ordering-reporting (fork) Iceberg build to exercise those assertions.
+
+ /** Cancels the test (see note above) unless the scan reported an ordering.
*/
+ private def assumeOrderingReported(plan: SparkPlan): Unit =
+ assume(
+ orderingReported(plan),
+ "current Iceberg build does not implement SupportsReportOrdering (no
ordering reported); " +
+ "sort-elimination assertion skipped")
+
+ // The reporting mechanism: SupportsReportOrdering ->
CometIcebergNativeScanExec.outputOrdering
+
+ test("native scan reports the table sort order for a multi-file sorted
table") {
+ withSortedTables(orderedReadConf)("t") { cat =>
+ spark.sql(s"CREATE TABLE $cat.db.t (id INT, data STRING) USING iceberg")
+ replaceSortOrder(cat, "db", "t", "id" -> true)
+ insertBatches(cat, "t", "(1,'a'),(3,'c')", "(2,'b'),(4,'d')")
+
+ val (_, plan) = checkSparkAnswer(s"SELECT id, data FROM $cat.db.t ORDER
BY id")
+ assume(nativeScans(plan).nonEmpty, "query did not use the native Iceberg
scan")
+ assumeOrderingReported(plan)
+ assert(
+
nativeScans(plan).head.outputOrdering.head.child.references.exists(_.name ==
"id"),
+ s"expected the scan to report an ordering on id:\n$plan")
+ }
+ }
+
+ test("native scan reports no ordering when the sort-merge flag is disabled")
{
+ withSortedTables(
+ orderedReadConf ++ Seq(CometConf.COMET_ICEBERG_SORT_MERGE_ENABLED.key ->
"false"))("t") {
+ cat =>
+ spark.sql(s"CREATE TABLE $cat.db.t (id INT, data STRING) USING
iceberg")
+ replaceSortOrder(cat, "db", "t", "id" -> true)
+ insertBatches(cat, "t", "(1,'a'),(3,'c')", "(2,'b'),(4,'d')")
+
+ // Correctness must hold with the feature off, and no ordering may be
advertised.
+ val (_, plan) = checkSparkAnswer(s"SELECT id, data FROM $cat.db.t
ORDER BY id")
+ nativeScans(plan).foreach { scan =>
+ assert(
+ scan.outputOrdering.isEmpty,
+ s"no ordering must be reported when the flag is off:\n$plan")
+ }
+ }
+ }
+
+ // K-way merge correctness (EnsureRequirements / EliminateSorts consume the
reported ordering;
+ // ORDER BY keeps the comparison order-sensitive so a merge defect is caught
directly).
+
+ test("merges multiple sorted files into one globally-ordered stream") {
+ withSortedTables(orderedReadConf)("t") { cat =>
+ spark.sql(s"CREATE TABLE $cat.db.t (id INT, data STRING) USING iceberg")
+ replaceSortOrder(cat, "db", "t", "id" -> true)
+ insertBatches(
+ cat,
+ "t",
+ "(1,'a'),(2,'b')",
+ "(3,'c'),(4,'d')",
+ "(5,'e'),(6,'f')",
+ "(7,'g'),(8,'h')")
+
+ val (_, plan) = checkSparkAnswer(s"SELECT id, data FROM $cat.db.t ORDER
BY id")
+ assert(nativeScans(plan).length == 1, s"expected exactly one native
scan:\n$plan")
+ }
+ }
+
+ test("merge interleaves duplicate sort-key values across files") {
+ withSortedTables(orderedReadConf)("t") { cat =>
+ spark.sql(s"CREATE TABLE $cat.db.t (id INT, data STRING) USING iceberg")
+ replaceSortOrder(cat, "db", "t", "id" -> true)
+ // The same id appears in several files; the merge must keep every row,
not drop or mis-order.
+ insertBatches(cat, "t", "(1,'a'),(2,'b')", "(1,'c'),(2,'d')",
"(1,'e'),(3,'f')")
+
+ checkSparkAnswer(s"SELECT id, data FROM $cat.db.t ORDER BY id, data")
+ }
+ }
+
+ test("merge applies merge-on-read deletes across a multi-file sorted
partition") {
+ withSortedTables(orderedReadConf)("t") { cat =>
+ spark.sql(
+ s"CREATE TABLE $cat.db.t (id INT, data STRING) USING iceberg " +
+ "TBLPROPERTIES ('format-version'='2',
'write.delete.mode'='merge-on-read')")
+ replaceSortOrder(cat, "db", "t", "id" -> true)
+ // Several files so a merge is required; then delete rows from some of
them. On a v2
+ // merge-on-read table DELETE writes delete files rather than rewriting
the data files, so
+ // the scan must apply the deletes while merging the still-sorted files.
+ insertBatches(cat, "t", "(1,'a'),(4,'d')", "(2,'b'),(5,'e')",
"(3,'c'),(6,'f')")
+ spark.sql(s"DELETE FROM $cat.db.t WHERE id IN (2, 5)")
+
+ checkSparkAnswer(s"SELECT id, data FROM $cat.db.t ORDER BY id")
+ }
+ }
+
+ test("merge honours NULLS FIRST on an ascending sort key") {
+ withSortedTables(spjConf)("t") { cat =>
+ spark.sql(
+ s"CREATE TABLE $cat.db.t (c1 INT, c2 STRING, c3 STRING) USING iceberg
" +
+ "PARTITIONED BY (c3)")
+ replaceSortOrder(cat, "db", "t", "c1" -> true) // ASC -> Iceberg default
NULLS FIRST
+ insertBatches(
+ cat,
+ "t",
+ "(null,'x','P1'),(3,'c','P1')",
+ "(null,'y','P1'),(1,'a','P1'),(2,'b','P1')")
+
+ checkSparkAnswer(
+ s"SELECT c1, c2 FROM $cat.db.t WHERE c3 = 'P1' ORDER BY c1 ASC NULLS
FIRST, c2")
+ }
+ }
+
+ test("merge honours NULLS LAST on a descending sort key") {
+ withSortedTables(spjConf)("t") { cat =>
+ spark.sql(
+ s"CREATE TABLE $cat.db.t (c1 INT, c2 STRING, c3 STRING) USING iceberg
" +
+ "PARTITIONED BY (c3)")
+ replaceSortOrder(cat, "db", "t", "c1" -> false) // DESC -> Iceberg
default NULLS LAST
+ insertBatches(
+ cat,
+ "t",
+ "(null,'x','P1'),(1,'a','P1')",
+ "(null,'y','P1'),(3,'c','P1'),(2,'b','P1')")
+
+ checkSparkAnswer(
+ s"SELECT c1, c2 FROM $cat.db.t WHERE c3 = 'P1' ORDER BY c1 DESC NULLS
LAST, c2")
+ }
+ }
+
+ test("merge on a descending sort order") {
+ withSortedTables(orderedReadConf)("t") { cat =>
+ spark.sql(s"CREATE TABLE $cat.db.t (id INT, data STRING) USING iceberg")
+ replaceSortOrder(cat, "db", "t", "id" -> false)
+ insertBatches(cat, "t", "(10,'j'),(9,'i')", "(8,'h'),(7,'g')",
"(6,'f'),(4,'d')")
+
+ checkSparkAnswer(s"SELECT id FROM $cat.db.t ORDER BY id DESC")
+ }
+ }
+
+ test("merge on a multi-column sort order") {
+ withSortedTables(orderedReadConf)("t") { cat =>
+ spark.sql(s"CREATE TABLE $cat.db.t (c1 INT, c2 STRING, c3 STRING) USING
iceberg")
+ replaceSortOrder(cat, "db", "t", "c3" -> true, "c1" -> true)
+ insertBatches(
+ cat,
+ "t",
+ "(1,'a','A'),(3,'c','A')",
+ "(2,'b','A'),(1,'a','B')",
+ "(2,'b','B'),(3,'c','B')")
+
+ checkSparkAnswer(s"SELECT c3, c1, c2 FROM $cat.db.t ORDER BY c3, c1")
+ }
+ }
+
+ test("single file needs no merge") {
+ withSortedTables(orderedReadConf)("t") { cat =>
+ spark.sql(s"CREATE TABLE $cat.db.t (id INT, data STRING) USING iceberg")
+ replaceSortOrder(cat, "db", "t", "id" -> true)
+ insertBatches(cat, "t", "(1,'a'),(2,'b')")
+
+ checkSparkAnswer(s"SELECT id, data FROM $cat.db.t ORDER BY id")
+ }
+ }
+
+ test("partitioned table with several files per partition") {
Review Comment:
Every test in this suite inserts 2 to 4 files per partition. Given the
memory concern in `iceberg_scan.rs` is specifically about a partition with a
large number of files (a live Parquet reader plus a buffered batch per file,
all opened at once), would it be worth adding one test that inserts something
like 50 to 100 single-row files into one partition? It would give the merge
path a correctness check under the actual shape of the workload this feature
targets, not just its happy-path shape, and would be a natural place to assert
on memory pool usage or peak concurrent readers once #5343 lands.
--
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]
