From 4515d705667580d877f2f0e3c651d5c4da895af6 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Thu, 1 Oct 2026 09:00:41 +0000 Subject: [PATCH 1/8] feat(scan): support validated FileScanTask ranges Preserve start and length in core and REST JSON, and reject partial reads until the reader can honor ranges. --- src/iceberg/catalog/rest/json_serde.cc | 54 ++++- src/iceberg/catalog/rest/types.cc | 1 + src/iceberg/data/file_scan_task_reader.cc | 6 + src/iceberg/data/file_scan_task_reader.h | 3 + src/iceberg/table_scan.cc | 71 ++++++- src/iceberg/table_scan.h | 34 +++ .../test/file_scan_task_reader_test.cc | 19 ++ src/iceberg/test/file_scan_task_test.cc | 115 +++++++++++ src/iceberg/test/rest_json_serde_test.cc | 193 ++++++++++++++++-- 9 files changed, 468 insertions(+), 28 deletions(-) diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc index ba348dec3f..d5a6ae6077 100644 --- a/src/iceberg/catalog/rest/json_serde.cc +++ b/src/iceberg/catalog/rest/json_serde.cc @@ -18,6 +18,7 @@ */ #include +#include #include #include #include @@ -110,6 +111,21 @@ constexpr std::string_view kDeleteFileReferences = "delete-file-references"; constexpr std::string_view kResidualFilter = "residual-filter"; constexpr std::string_view kStart = "start"; constexpr std::string_view kLength = "length"; +constexpr std::string_view kMapKeys = "keys"; +constexpr std::string_view kMapValues = "values"; + +Result GetRangeValue(const nlohmann::json& json, std::string_view key) { + const auto& value = json.at(key); + if (!value.is_number_integer()) { + return JsonParseError("'{}' must be an integer, but is {}", key, value.type_name()); + } + if (value.is_number_unsigned() && + value.get() > + static_cast(std::numeric_limits::max())) { + return JsonParseError("'{}' integer is out of range: {}", key, SafeDumpJson(value)); + } + return value.get(); +} Result StorageCredentialToJson(const StorageCredential& credential) { ICEBERG_RETURN_UNEXPECTED(credential.Validate()); @@ -168,14 +184,18 @@ Result>> FileScanTasksFromJson( ICEBERG_ASSIGN_OR_RAISE(auto data_file_json, GetJsonValue(task_json, kDataFile)); ICEBERG_ASSIGN_OR_RAISE( - auto data_file, - iceberg::DataFileFromJson(data_file_json, partition_spec_by_id, schema)); - ICEBERG_ASSIGN_OR_RAISE(auto start, GetJsonIntegerOrDefault(task_json, kStart, 0)); - ICEBERG_ASSIGN_OR_RAISE( - auto length, - GetJsonIntegerOrDefault(task_json, kLength, data_file.file_size_in_bytes)); - ICEBERG_RETURN_UNEXPECTED( - CheckFileScanTaskNotSplit(start, length, data_file.file_size_in_bytes)); + auto data_file, DataFileFromJson(data_file_json, partition_spec_by_id, schema)); + const bool has_start = task_json.contains(kStart); + const bool has_length = task_json.contains(kLength); + if (has_start != has_length) { + return JsonParseError("'start' and 'length' must be provided together"); + } + int64_t start = 0; + int64_t length = data_file.file_size_in_bytes; + if (has_start) { + ICEBERG_ASSIGN_OR_RAISE(start, GetRangeValue(task_json, kStart)); + ICEBERG_ASSIGN_OR_RAISE(length, GetRangeValue(task_json, kLength)); + } // FIXME: REST scan-task DataFile JSON currently carries first-row-id, // but not the manifest-entry data sequence number. Until the REST API exposes // it, REST-planned tasks cannot inherit _last_updated_sequence_number. @@ -203,9 +223,13 @@ Result>> FileScanTasksFromJson( ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json)); } - file_scan_tasks.push_back(std::make_shared( - std::make_shared(std::move(data_file)), std::move(task_delete_files), - std::move(residual_filter))); + auto task = FileScanTask::MakeSplit(std::make_shared(std::move(data_file)), + start, length, std::move(task_delete_files), + std::move(residual_filter)); + if (!task) { + return JsonParseError("Invalid FileScanTask range: {}", task.error().message); + } + file_scan_tasks.push_back(std::move(task.value())); } return file_scan_tasks; } @@ -253,10 +277,18 @@ Result ScanTaskFieldsToJson( if (!task) continue; nlohmann::json task_json; if (task->data_file()) { + auto range_result = + FileScanTask::MakeSplit(task->data_file(), task->start(), task->length()); + if (!range_result) { + return ValidationFailed("Invalid FileScanTask range: {}", + range_result.error().message); + } ICEBERG_ASSIGN_OR_RAISE( auto data_file_json, iceberg::ToJson(*task->data_file(), partition_specs_by_id, schema)); task_json[kDataFile] = std::move(data_file_json); + task_json[kStart] = task->start(); + task_json[kLength] = task->length(); } if (!task->delete_files().empty()) { std::vector refs; diff --git a/src/iceberg/catalog/rest/types.cc b/src/iceberg/catalog/rest/types.cc index 84fba9a7ca..ebc2c2876f 100644 --- a/src/iceberg/catalog/rest/types.cc +++ b/src/iceberg/catalog/rest/types.cc @@ -192,6 +192,7 @@ bool FileScanTaskEqual(const std::shared_ptr& lhs, return false; } return SharedPtrEqual(lhs->data_file(), rhs->data_file()) && + lhs->start() == rhs->start() && lhs->length() == rhs->length() && SharedPtrVectorEqual(lhs->delete_files(), rhs->delete_files()) && ExpressionPtrEqual(lhs->residual_filter(), rhs->residual_filter()); } diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 0c41f25eda..7d3a8bc7e3 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -163,6 +163,12 @@ class FileScanTaskReader::Impl { ICEBERG_PRECHECK(data_file->file_size_in_bytes >= 0, "Data file size must not be negative: {}", data_file->file_size_in_bytes); + if (task.is_split()) { + return NotSupported( + "Reading a partial FileScanTask is not supported yet: {} " + "(start={}, length={})", + data_file->file_path, task.start(), task.length()); + } if (task.delete_files().empty()) { auto options = MakeReaderOptions( diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h index a71ef5f84f..c34f3e7aa8 100644 --- a/src/iceberg/data/file_scan_task_reader.h +++ b/src/iceberg/data/file_scan_task_reader.h @@ -67,6 +67,9 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader { ~FileScanTaskReader(); /// \brief Open a task and return an Arrow C stream for its projected live rows. + /// + /// Partial tasks are rejected until range-aware reading is implemented. The task and + /// its shared data/delete-file metadata must not be mutated while the stream uses them. Result Open(const FileScanTask& task); FileScanTaskReader(const FileScanTaskReader&) = delete; diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index 46f235fbe2..c4df36f366 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -37,6 +37,8 @@ #include "iceberg/snapshot.h" #include "iceberg/table.h" #include "iceberg/table_metadata.h" +#include "iceberg/util/content_file_util.h" +#include "iceberg/util/int128.h" #include "iceberg/util/macros.h" #include "iceberg/util/snapshot_util.h" #include "iceberg/util/timepoint.h" @@ -179,11 +181,76 @@ FileScanTask::FileScanTask(std::shared_ptr data_file, ICEBERG_DCHECK(data_file_ != nullptr, "Data file cannot be null for FileScanTask"); } -int64_t FileScanTask::size_bytes() const { return data_file_->file_size_in_bytes; } +FileScanTask::FileScanTask(std::shared_ptr data_file, + std::vector> delete_files, + std::shared_ptr residual_filter, Range range) + : FileScanTask(std::move(data_file), std::move(delete_files), + std::move(residual_filter)) { + range_ = range; +} + +Result> FileScanTask::MakeSplit( + std::shared_ptr data_file, int64_t start, int64_t length, + std::vector> delete_files, + std::shared_ptr filter) { + if (data_file == nullptr) { + return InvalidArgument("Cannot create a file scan task without a data file"); + } + + const int64_t file_size = data_file->file_size_in_bytes; + if (file_size < 0) { + return InvalidArgument("Cannot split file {} with negative size {}", + data_file->file_path, file_size); + } + if (start < 0 || start > file_size) { + return InvalidArgument("Split start {} is outside file {} of size {}", start, + data_file->file_path, file_size); + } + // Subtraction is safe after validating start, even when file_size is INT64_MAX. + if (length < 0 || length > file_size - start) { + return InvalidArgument("Split length {} from start {} is outside file {} of size {}", + length, start, data_file->file_path, file_size); + } + if (start == 0 && length == file_size) { + return std::make_shared(std::move(data_file), std::move(delete_files), + std::move(filter)); + } + if (length == 0) { + return InvalidArgument("A partial split of file {} must have positive length", + data_file->file_path); + } + + return std::shared_ptr( + new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter), + Range{start, length, file_size})); +} + +int64_t FileScanTask::length() const { + return range_ ? range_->length : data_file_->file_size_in_bytes; +} + +int64_t FileScanTask::size_bytes() const { return length(); } int32_t FileScanTask::files_count() const { return 1; } -int64_t FileScanTask::estimated_row_count() const { return data_file_->record_count; } +int64_t FileScanTask::estimated_row_count() const { + if (!range_) { + return data_file_->record_count; + } + + const int64_t record_count = data_file_->record_count; + if (record_count <= 0) { + return 0; + } + + // Cumulative boundaries make estimates additive across contiguous splits, and + // 128-bit products avoid overflow for any valid int64 file size and row count. + const auto end = static_cast(range_->start) + range_->length; + const auto count_at_end = end * record_count / range_->file_size; + const auto count_at_start = + static_cast(range_->start) * record_count / range_->file_size; + return static_cast(count_at_end - count_at_start); +} // ChangelogScanTask implementation diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h index 6ec29ad9f6..b6137d549e 100644 --- a/src/iceberg/table_scan.h +++ b/src/iceberg/table_scan.h @@ -22,6 +22,7 @@ /// \file iceberg/table_scan.h /// \brief Define table scan APIs and scan task types. +#include #include #include #include @@ -71,10 +72,32 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask { /// \param data_file The data file to read. /// \param delete_files Delete files that apply to this data file. /// \param filter Optional residual filter to apply after reading. + /// \note The data file, delete files, and residual filter must not be mutated while + /// a planning or reading stream uses this task. explicit FileScanTask(std::shared_ptr data_file, std::vector> delete_files = {}, std::shared_ptr filter = nullptr); + /// \brief Make a task for a validated half-open byte range [start, start + length). + /// + /// A range covering the entire file is represented as a whole-file task. An empty + /// range is valid only for an empty whole file. The data file, delete files, and + /// residual filter are shared with the resulting task and must not be mutated + /// while a planning or reading stream uses it. + static Result> MakeSplit( + std::shared_ptr data_file, int64_t start, int64_t length, + std::vector> delete_files = {}, + std::shared_ptr filter = nullptr); + + /// \brief The inclusive byte offset where this task starts, or zero for a whole file. + int64_t start() const { return range_ ? range_->start : 0; } + + /// \brief The byte length of this task, or the current size of a whole file. + int64_t length() const; + + /// \brief Whether this task reads a proper subset of its data file. + bool is_split() const { return range_.has_value(); } + /// \brief The data file that should be read by this scan task. const std::shared_ptr& data_file() const { return data_file_; } @@ -92,9 +115,20 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask { int64_t estimated_row_count() const override; private: + struct Range { + int64_t start; + int64_t length; + int64_t file_size; + }; + + FileScanTask(std::shared_ptr data_file, + std::vector> delete_files, + std::shared_ptr filter, Range range); + std::shared_ptr data_file_; std::vector> delete_files_; std::shared_ptr residual_filter_; + std::optional range_; }; /// \brief Stream of file scan tasks. diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index cca13ce0df..11366f72d9 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -409,6 +409,25 @@ TEST_F(FileScanTaskReaderTest, OpenWithoutDeletesReadsProjectedSchema) { VerifyStream(&stream, R"([[1, "Foo"], [2, "Bar"], [3, "Baz"]])")); } +TEST_F(FileScanTaskReaderTest, RejectsPartialTaskUntilRangeReadsAreSupported) { + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, + MakeDataFile(table_schema_, + R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz", "green"]])")); + ICEBERG_UNWRAP_OR_FAIL(auto task, FileScanTask::MakeSplit(data_file, 0, 1)); + + FileScanTaskReader::Options options{ + .io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = projected_schema_, + }; + ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options))); + auto result = reader->Open(*task); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); + EXPECT_THAT(result, HasErrorMessage("Reading a partial FileScanTask")); +} + TEST_F(FileScanTaskReaderTest, ReadLastUpdatedFromDataSeq) { ICEBERG_UNWRAP_OR_FAIL( auto data_file, diff --git a/src/iceberg/test/file_scan_task_test.cc b/src/iceberg/test/file_scan_task_test.cc index cb945ba1fb..6394c96fed 100644 --- a/src/iceberg/test/file_scan_task_test.cc +++ b/src/iceberg/test/file_scan_task_test.cc @@ -17,6 +17,10 @@ * under the License. */ +#include +#include +#include + #include #include #include @@ -29,6 +33,7 @@ #include "iceberg/arrow/arrow_io_internal.h" #include "iceberg/data/file_scan_task_reader.h" +#include "iceberg/expression/expressions.h" #include "iceberg/file_format.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/parquet/parquet_register.h" @@ -40,6 +45,116 @@ namespace iceberg { +TEST(FileScanTaskRangeTest, WholeFileAndEmptyFile) { + auto data_file = std::make_shared(); + data_file->file_path = "test.parquet"; + data_file->file_size_in_bytes = 100; + data_file->record_count = 7; + + FileScanTask whole(data_file); + EXPECT_FALSE(whole.is_split()); + EXPECT_EQ(whole.start(), 0); + EXPECT_EQ(whole.length(), 100); + EXPECT_EQ(whole.size_bytes(), 100); + EXPECT_EQ(whole.estimated_row_count(), 7); + + ICEBERG_UNWRAP_OR_FAIL(auto explicit_whole, FileScanTask::MakeSplit(data_file, 0, 100)); + EXPECT_FALSE(explicit_whole->is_split()); + EXPECT_EQ(explicit_whole->start(), 0); + EXPECT_EQ(explicit_whole->length(), 100); + EXPECT_EQ(explicit_whole->estimated_row_count(), 7); + + data_file->file_size_in_bytes = 0; + data_file->record_count = 0; + ICEBERG_UNWRAP_OR_FAIL(auto empty, FileScanTask::MakeSplit(data_file, 0, 0)); + EXPECT_FALSE(empty->is_split()); + EXPECT_EQ(empty->start(), 0); + EXPECT_EQ(empty->length(), 0); + EXPECT_EQ(empty->size_bytes(), 0); + EXPECT_EQ(empty->estimated_row_count(), 0); +} + +TEST(FileScanTaskRangeTest, DistinctSplitsShareMetadataAndEstimateRowsOnce) { + auto data_file = std::make_shared(); + data_file->file_path = "test.parquet"; + data_file->file_size_in_bytes = 100; + data_file->record_count = 7; + data_file->first_row_id = 1000; + data_file->data_sequence_number = 22; + auto delete_file = std::make_shared(); + auto residual = Expressions::AlwaysTrue(); + + ICEBERG_UNWRAP_OR_FAIL( + auto first, FileScanTask::MakeSplit(data_file, 0, 40, {delete_file}, residual)); + ICEBERG_UNWRAP_OR_FAIL( + auto second, FileScanTask::MakeSplit(data_file, 40, 60, {delete_file}, residual)); + + EXPECT_TRUE(first->is_split()); + EXPECT_TRUE(second->is_split()); + EXPECT_EQ(first->start(), 0); + EXPECT_EQ(first->length(), 40); + EXPECT_EQ(second->start(), 40); + EXPECT_EQ(second->length(), 60); + EXPECT_EQ(first->size_bytes() + second->size_bytes(), 100); + EXPECT_EQ(first->estimated_row_count(), 2); + EXPECT_EQ(second->estimated_row_count(), 5); + EXPECT_EQ(first->estimated_row_count() + second->estimated_row_count(), 7); + EXPECT_EQ(first->data_file(), data_file); + EXPECT_EQ(second->data_file(), data_file); + ASSERT_EQ(first->delete_files().size(), 1); + ASSERT_EQ(second->delete_files().size(), 1); + EXPECT_EQ(first->delete_files()[0], delete_file); + EXPECT_EQ(second->delete_files()[0], delete_file); + EXPECT_EQ(first->residual_filter(), residual); + EXPECT_EQ(second->residual_filter(), residual); + EXPECT_EQ(first->data_file()->first_row_id, 1000); + EXPECT_EQ(first->data_file()->data_sequence_number, 22); + EXPECT_EQ(second->data_file()->first_row_id, 1000); + EXPECT_EQ(second->data_file()->data_sequence_number, 22); +} + +TEST(FileScanTaskRangeTest, RejectsInvalidRanges) { + auto data_file = std::make_shared(); + data_file->file_path = "test.parquet"; + data_file->file_size_in_bytes = 100; + + const auto expect_invalid = [&](int64_t start, int64_t length) { + auto result = FileScanTask::MakeSplit(data_file, start, length); + ASSERT_FALSE(result.has_value()); + EXPECT_EQ(result.error().kind, ErrorKind::kInvalidArgument); + }; + expect_invalid(-1, 1); + expect_invalid(0, -1); + expect_invalid(0, 0); + expect_invalid(100, 0); + expect_invalid(101, 1); + expect_invalid(99, 2); + + auto null_file = FileScanTask::MakeSplit(nullptr, 0, 1); + ASSERT_FALSE(null_file.has_value()); + EXPECT_EQ(null_file.error().kind, ErrorKind::kInvalidArgument); + + data_file->file_size_in_bytes = -1; + expect_invalid(0, 1); + + data_file->file_size_in_bytes = std::numeric_limits::max(); + expect_invalid(std::numeric_limits::max() - 1, 2); +} + +TEST(FileScanTaskRangeTest, RowEstimateDoesNotOverflow) { + auto data_file = std::make_shared(); + data_file->file_size_in_bytes = std::numeric_limits::max(); + data_file->record_count = std::numeric_limits::max(); + + ICEBERG_UNWRAP_OR_FAIL( + auto first, + FileScanTask::MakeSplit(data_file, 0, data_file->file_size_in_bytes - 1)); + ICEBERG_UNWRAP_OR_FAIL(auto last, FileScanTask::MakeSplit( + data_file, data_file->file_size_in_bytes - 1, 1)); + EXPECT_EQ(first->estimated_row_count(), data_file->record_count - 1); + EXPECT_EQ(last->estimated_row_count(), 1); +} + class FileScanTaskTest : public TempFileTestBase { protected: static void SetUpTestSuite() { parquet::RegisterAll(); } diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc index 34ca4a45b3..39427f293d 100644 --- a/src/iceberg/test/rest_json_serde_test.cc +++ b/src/iceberg/test/rest_json_serde_test.cc @@ -17,6 +17,7 @@ * under the License. */ +#include #include #include #include @@ -2127,6 +2128,9 @@ TEST(FileScanTasksFromJsonTest, SingleTaskNoDeleteFiles) { const auto& task = result.value()[0]; ASSERT_NE(task->data_file(), nullptr); EXPECT_EQ(task->data_file()->file_path, "s3://bucket/data/file.parquet"); + EXPECT_EQ(task->start(), 0); + EXPECT_EQ(task->length(), 12345); + EXPECT_FALSE(task->is_split()); EXPECT_TRUE(task->delete_files().empty()); EXPECT_EQ(task->residual_filter(), nullptr); } @@ -2146,11 +2150,15 @@ TEST(FileScanTasksFromJsonTest, AcceptsWholeFileStartAndLength) { "length": 12345 }])"_json; - EXPECT_THAT(FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0)), - IsOk()); + auto result = FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0)); + ASSERT_THAT(result, IsOk()); + ASSERT_EQ(result.value().size(), 1U); + EXPECT_EQ(result.value()[0]->start(), 0); + EXPECT_EQ(result.value()[0]->length(), 12345); + EXPECT_FALSE(result.value()[0]->is_split()); } -TEST(FileScanTasksFromJsonTest, NullStartAndLengthUseWholeFileDefaults) { +TEST(FileScanTasksFromJsonTest, RejectsNullOrMissingRangeEndpoint) { auto json = R"([{ "data-file": { "content": "data", @@ -2161,12 +2169,25 @@ TEST(FileScanTasksFromJsonTest, NullStartAndLengthUseWholeFileDefaults) { "file-size-in-bytes": 12345, "record-count": 100 }, - "start": null, - "length": null + "start": 0, + "length": 12345 }])"_json; - EXPECT_THAT(FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0)), - IsOk()); + for (std::string_view missing_field : {"start", "length"}) { + SCOPED_TRACE(missing_field); + auto incomplete_json = json; + incomplete_json[0].erase(std::string(missing_field)); + auto result = + FileScanTasksFromJson(incomplete_json, {}, UnpartitionedSpecs(), Schema({}, 0)); + EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(result, HasErrorMessage("must be provided together")); + } + + json[0]["start"] = nullptr; + json[0]["length"] = nullptr; + auto null_result = FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0)); + EXPECT_THAT(null_result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(null_result, HasErrorMessage("'start' must be an integer")); } TEST(FileScanTasksFromJsonTest, RejectsInvalidRangeIntegers) { @@ -2202,7 +2223,7 @@ TEST(FileScanTasksFromJsonTest, RejectsInvalidRangeIntegers) { } } -TEST(FileScanTasksFromJsonTest, RejectsSplitTaskWhenEitherRangeFieldDiffers) { +TEST(FileScanTasksFromJsonTest, AcceptsValidPartialRanges) { auto json = R"([{ "data-file": { "content": "data", @@ -2217,17 +2238,85 @@ TEST(FileScanTasksFromJsonTest, RejectsSplitTaskWhenEitherRangeFieldDiffers) { "length": 12345 }])"_json; - for (const auto& [field, value] : - {std::pair{"start", 100}, - std::pair{"length", 200}}) { - SCOPED_TRACE(field); + for (const auto& [start, length] : + {std::pair{0, 200}, {100, 200}, {12245, 100}}) { + SCOPED_TRACE(testing::Message() << "start=" << start << ", length=" << length); auto split_json = json; - split_json[0][field] = value; + split_json[0]["start"] = start; + split_json[0]["length"] = length; auto result = FileScanTasksFromJson(split_json, {}, UnpartitionedSpecs(), Schema({}, 0)); - EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); - EXPECT_THAT(result, HasErrorMessage("Split FileScanTask is not supported")); + ASSERT_THAT(result, IsOk()); + ASSERT_EQ(result.value().size(), 1U); + EXPECT_EQ(result.value()[0]->start(), start); + EXPECT_EQ(result.value()[0]->length(), length); + EXPECT_TRUE(result.value()[0]->is_split()); + } +} + +TEST(FileScanTasksFromJsonTest, RejectsInvalidRanges) { + auto json = R"([{ + "data-file": { + "content": "data", + "file-path": "s3://bucket/data/file.parquet", + "file-format": "PARQUET", + "spec-id": 0, + "partition": [], + "file-size-in-bytes": 12345, + "record-count": 100 + }, + "start": 0, + "length": 12345 + }])"_json; + + for (const auto& [start, length] : {std::pair{-1, 100}, + {0, -1}, + {0, 0}, + {12345, 0}, + {12345, 1}, + {12000, 500}}) { + SCOPED_TRACE(testing::Message() << "start=" << start << ", length=" << length); + auto invalid_json = json; + invalid_json[0]["start"] = start; + invalid_json[0]["length"] = length; + auto result = + FileScanTasksFromJson(invalid_json, {}, UnpartitionedSpecs(), Schema({}, 0)); + EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(result, HasErrorMessage("Invalid FileScanTask range")); } + + auto overflow_json = json; + overflow_json[0]["data-file"]["file-size-in-bytes"] = + std::numeric_limits::max(); + overflow_json[0]["start"] = std::numeric_limits::max() - 1; + overflow_json[0]["length"] = 2; + auto overflow_result = + FileScanTasksFromJson(overflow_json, {}, UnpartitionedSpecs(), Schema({}, 0)); + EXPECT_THAT(overflow_result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(overflow_result, HasErrorMessage("Invalid FileScanTask range")); +} + +TEST(FileScanTasksFromJsonTest, AcceptsEmptyWholeFile) { + auto json = R"([{ + "data-file": { + "content": "data", + "file-path": "s3://bucket/data/empty.parquet", + "file-format": "PARQUET", + "spec-id": 0, + "partition": [], + "file-size-in-bytes": 0, + "record-count": 0 + }, + "start": 0, + "length": 0 + }])"_json; + + auto result = FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0)); + ASSERT_THAT(result, IsOk()); + ASSERT_EQ(result.value().size(), 1U); + EXPECT_EQ(result.value()[0]->start(), 0); + EXPECT_EQ(result.value()[0]->length(), 0); + EXPECT_FALSE(result.value()[0]->is_split()); } TEST(FileScanTasksFromJsonTest, RowLineageSequence) { @@ -2400,6 +2489,80 @@ TEST(FetchScanTasksResponseRoundtripTest, WithFileScanTasksAndDeleteFiles) { EXPECT_EQ(*result, *result2); } +TEST(FetchScanTasksResponseRoundtripTest, PreservesDistinctSplitRanges) { + auto json = nlohmann::json::parse(R"({ + "file-scan-tasks": [ + { + "data-file": { + "content": "data", + "file-path": "s3://bucket/data/file.parquet", + "file-format": "PARQUET", + "spec-id": 0, + "partition": [], + "file-size-in-bytes": 12345, + "record-count": 100 + }, + "start": 0, + "length": 5000 + }, + { + "data-file": { + "content": "data", + "file-path": "s3://bucket/data/file.parquet", + "file-format": "PARQUET", + "spec-id": 0, + "partition": [], + "file-size-in-bytes": 12345, + "record-count": 100 + }, + "start": 5000, + "length": 7345 + } + ] + })"); + + auto result = FetchScanTasksResponseFromJson(json, UnpartitionedSpecs(), EmptySchema()); + ASSERT_THAT(result, IsOk()); + ASSERT_TRUE(result.value().file_scan_tasks.has_value()); + ASSERT_EQ(result.value().file_scan_tasks->size(), 2U); + EXPECT_EQ(result.value().file_scan_tasks->at(0)->start(), 0); + EXPECT_EQ(result.value().file_scan_tasks->at(0)->length(), 5000); + EXPECT_EQ(result.value().file_scan_tasks->at(1)->start(), 5000); + EXPECT_EQ(result.value().file_scan_tasks->at(1)->length(), 7345); + + ICEBERG_UNWRAP_OR_FAIL(auto roundtrip_json, + ToJson(result.value(), UnpartitionedSpecs(), EmptySchema())); + EXPECT_EQ(roundtrip_json["file-scan-tasks"][0]["start"], 0); + EXPECT_EQ(roundtrip_json["file-scan-tasks"][0]["length"], 5000); + EXPECT_EQ(roundtrip_json["file-scan-tasks"][1]["start"], 5000); + EXPECT_EQ(roundtrip_json["file-scan-tasks"][1]["length"], 7345); + + auto result2 = + FetchScanTasksResponseFromJson(roundtrip_json, UnpartitionedSpecs(), EmptySchema()); + ASSERT_THAT(result2, IsOk()); + EXPECT_EQ(result.value(), result2.value()); +} + +TEST(FetchScanTasksResponseRoundtripTest, RejectsSplitOutsideChangedDataFile) { + auto data_file = std::make_shared(); + data_file->content = DataFile::Content::kData; + data_file->file_path = "s3://bucket/data/file.parquet"; + data_file->file_format = FileFormatType::kParquet; + data_file->partition_spec_id = PartitionSpec::kInitialSpecId; + data_file->partition = PartitionValues{}; + data_file->file_size_in_bytes = 12345; + data_file->record_count = 100; + ICEBERG_UNWRAP_OR_FAIL(auto split, FileScanTask::MakeSplit(data_file, 5000, 7345)); + + FetchScanTasksResponse response; + response.file_scan_tasks = std::vector>{split}; + data_file->file_size_in_bytes = 12000; + + auto result = ToJson(response, UnpartitionedSpecs(), EmptySchema()); + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT(result, HasErrorMessage("Invalid FileScanTask range")); +} + TEST(FetchScanTasksResponseRoundtripTest, PreservesResidualFilter) { auto json = nlohmann::json::parse(R"({ "file-scan-tasks": [ From 3bae712cf09a42ab2868aff408a1ceca4384a0ac Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Sun, 4 Oct 2026 04:39:50 +0000 Subject: [PATCH 2/8] fix(scan): use designated initializer for task range --- src/iceberg/table_scan.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index c4df36f366..51f65fa0e9 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -222,7 +222,7 @@ Result> FileScanTask::MakeSplit( return std::shared_ptr( new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter), - Range{start, length, file_size})); + Range{.start = start, .length = length, .file_size = file_size})); } int64_t FileScanTask::length() const { From 7fcb4fd104294e72803da1325451e265979ad593 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Mon, 12 Oct 2026 00:21:46 +0000 Subject: [PATCH 3/8] docs(scan): explain FileScanTask range logic Co-authored-by: Isaac --- src/iceberg/catalog/rest/json_serde.cc | 3 +++ src/iceberg/table_scan.cc | 6 ++++-- src/iceberg/table_scan.h | 2 ++ 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc index d5a6ae6077..fafcc50c6b 100644 --- a/src/iceberg/catalog/rest/json_serde.cc +++ b/src/iceberg/catalog/rest/json_serde.cc @@ -114,6 +114,7 @@ constexpr std::string_view kLength = "length"; constexpr std::string_view kMapKeys = "keys"; constexpr std::string_view kMapValues = "values"; +// Reject values above INT64_MAX; get() would wrap them to negative. Result GetRangeValue(const nlohmann::json& json, std::string_view key) { const auto& value = json.at(key); if (!value.is_number_integer()) { @@ -185,6 +186,7 @@ Result>> FileScanTasksFromJson( GetJsonValue(task_json, kDataFile)); ICEBERG_ASSIGN_OR_RAISE( auto data_file, DataFileFromJson(data_file_json, partition_spec_by_id, schema)); + // Neither field means the whole file. Only one field is ambiguous. const bool has_start = task_json.contains(kStart); const bool has_length = task_json.contains(kLength); if (has_start != has_length) { @@ -277,6 +279,7 @@ Result ScanTaskFieldsToJson( if (!task) continue; nlohmann::json task_json; if (task->data_file()) { + // The shared data file may have changed size; don't write an invalid range. auto range_result = FileScanTask::MakeSplit(task->data_file(), task->start(), task->length()); if (!range_result) { diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index 51f65fa0e9..b83599e2c4 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -211,6 +211,8 @@ Result> FileScanTask::MakeSplit( return InvalidArgument("Split length {} from start {} is outside file {} of size {}", length, start, data_file->file_path, file_size); } + // A full-file range is a whole-file task (is_split() is false). This is also the + // only valid empty range: [0, 0) of an empty file. if (start == 0 && length == file_size) { return std::make_shared(std::move(data_file), std::move(delete_files), std::move(filter)); @@ -243,8 +245,8 @@ int64_t FileScanTask::estimated_row_count() const { return 0; } - // Cumulative boundaries make estimates additive across contiguous splits, and - // 128-bit products avoid overflow for any valid int64 file size and row count. + // floor(end * rows / size) - floor(start * rows / size), so splits that cover + // the file sum exactly to record_count. int128 avoids overflow. const auto end = static_cast(range_->start) + range_->length; const auto count_at_end = end * record_count / range_->file_size; const auto count_at_start = diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h index b6137d549e..3d4893e758 100644 --- a/src/iceberg/table_scan.h +++ b/src/iceberg/table_scan.h @@ -115,9 +115,11 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask { int64_t estimated_row_count() const override; private: + // Only set for a partial file; whole-file tasks leave range_ empty. struct Range { int64_t start; int64_t length; + // Size when the range was validated. Always positive, so safe to divide by. int64_t file_size; }; From 10e96001f344c37df6f9e95364468376fc434ce4 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Tue, 13 Oct 2026 07:00:26 +0000 Subject: [PATCH 4/8] fix comments --- src/iceberg/catalog/rest/json_serde.cc | 18 +---- src/iceberg/json_serde.cc | 25 ++----- src/iceberg/json_serde_internal.h | 4 -- src/iceberg/test/json_serde_test.cc | 86 +++++++++++++++++++++--- src/iceberg/test/rest_json_serde_test.cc | 2 +- 5 files changed, 85 insertions(+), 50 deletions(-) diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc index fafcc50c6b..e6cdab23ff 100644 --- a/src/iceberg/catalog/rest/json_serde.cc +++ b/src/iceberg/catalog/rest/json_serde.cc @@ -114,20 +114,6 @@ constexpr std::string_view kLength = "length"; constexpr std::string_view kMapKeys = "keys"; constexpr std::string_view kMapValues = "values"; -// Reject values above INT64_MAX; get() would wrap them to negative. -Result GetRangeValue(const nlohmann::json& json, std::string_view key) { - const auto& value = json.at(key); - if (!value.is_number_integer()) { - return JsonParseError("'{}' must be an integer, but is {}", key, value.type_name()); - } - if (value.is_number_unsigned() && - value.get() > - static_cast(std::numeric_limits::max())) { - return JsonParseError("'{}' integer is out of range: {}", key, SafeDumpJson(value)); - } - return value.get(); -} - Result StorageCredentialToJson(const StorageCredential& credential) { ICEBERG_RETURN_UNEXPECTED(credential.Validate()); nlohmann::json json; @@ -195,8 +181,8 @@ Result>> FileScanTasksFromJson( int64_t start = 0; int64_t length = data_file.file_size_in_bytes; if (has_start) { - ICEBERG_ASSIGN_OR_RAISE(start, GetRangeValue(task_json, kStart)); - ICEBERG_ASSIGN_OR_RAISE(length, GetRangeValue(task_json, kLength)); + ICEBERG_ASSIGN_OR_RAISE(start, GetJsonInteger(task_json, kStart)); + ICEBERG_ASSIGN_OR_RAISE(length, GetJsonInteger(task_json, kLength)); } // FIXME: REST scan-task DataFile JSON currently carries first-row-id, // but not the manifest-entry data sequence number. Until the REST API exposes diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc index c1c3ffb73f..d64099bfd9 100644 --- a/src/iceberg/json_serde.cc +++ b/src/iceberg/json_serde.cc @@ -2423,8 +2423,8 @@ Result ToJson( } json[kSpec] = ToJson(*spec_it->second); json[kDataFile] = std::move(data_file_json); - json[kStart] = 0; - json[kLength] = task.data_file()->file_size_in_bytes; + json[kStart] = task.start(); + json[kLength] = task.length(); nlohmann::json delete_files_json = nlohmann::json::array(); for (const auto& delete_file : task.delete_files()) { @@ -2457,17 +2457,6 @@ Result ToJson( return json; } -Status CheckFileScanTaskNotSplit(int64_t start, int64_t length, - int64_t file_size_in_bytes) { - if (start != 0 || length != file_size_in_bytes) { - return NotSupported( - "Split FileScanTask is not supported: start={}, length={}, " - "file-size-in-bytes={}", - start, length, file_size_in_bytes); - } - return {}; -} - Result> FileScanTaskFromJson(const nlohmann::json& json) { if (!json.is_object()) { return JsonParseError("Cannot parse file scan task from a non-object: {}", @@ -2510,8 +2499,6 @@ Result> FileScanTaskFromJson(const nlohmann::json& ICEBERG_ASSIGN_OR_RAISE(auto start, GetJsonInteger(json, kStart)); ICEBERG_ASSIGN_OR_RAISE(auto length, GetJsonInteger(json, kLength)); - ICEBERG_RETURN_UNEXPECTED( - CheckFileScanTaskNotSplit(start, length, data_file.file_size_in_bytes)); std::vector> delete_files; if (json.contains(kDeleteFiles)) { @@ -2539,9 +2526,11 @@ Result> FileScanTaskFromJson(const nlohmann::json& ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json)); } - return std::make_shared(std::make_shared(std::move(data_file)), - std::move(delete_files), - std::move(residual_filter)); + ICEBERG_ASSIGN_OR_RAISE(auto task, FileScanTask::MakeSplit( + std::make_shared(std::move(data_file)), start, length, + std::move(delete_files), std::move(residual_filter))); + + return std::move(task); } } // namespace iceberg diff --git a/src/iceberg/json_serde_internal.h b/src/iceberg/json_serde_internal.h index 2465df1404..1d7e06ea87 100644 --- a/src/iceberg/json_serde_internal.h +++ b/src/iceberg/json_serde_internal.h @@ -474,8 +474,4 @@ ICEBERG_EXPORT Result ToJson( ICEBERG_EXPORT Result> FileScanTaskFromJson( const nlohmann::json& json); -/// Rejects `start != 0 || length != file_size_in_bytes`. -ICEBERG_EXPORT Status CheckFileScanTaskNotSplit(int64_t start, int64_t length, - int64_t file_size_in_bytes); - } // namespace iceberg diff --git a/src/iceberg/test/json_serde_test.cc b/src/iceberg/test/json_serde_test.cc index 51bc2af11c..7e95eb2ae0 100644 --- a/src/iceberg/test/json_serde_test.cc +++ b/src/iceberg/test/json_serde_test.cc @@ -18,6 +18,7 @@ */ #include +#include #include #include #include @@ -1803,31 +1804,94 @@ TEST_F(FileScanTaskJsonTest, RequiresJavaCoreFields) { } } -TEST_F(FileScanTaskJsonTest, RejectsSplitTasksUsingExactOrCondition) { +TEST_F(FileScanTaskJsonTest, RoundTripsValidRanges) { struct Case { int64_t start; int64_t length; - bool supported; + bool is_split; }; - for (const auto& test_case : {Case{.start = 0, .length = 10, .supported = true}, - Case{.start = 1, .length = 10, .supported = false}, - Case{.start = 0, .length = 9, .supported = false}}) { + for (const auto& test_case : {Case{.start = 0, .length = 10, .is_split = false}, + Case{.start = 0, .length = 9, .is_split = true}, + Case{.start = 4, .length = 6, .is_split = true}}) { SCOPED_TRACE(testing::Message() << "start=" << test_case.start << ", length=" << test_case.length); auto json = JavaGolden(); json["start"] = test_case.start; json["length"] = test_case.length; + json["residual-filter"] = R"({"type":"gt","term":"id","value":5})"_json; + + ICEBERG_UNWRAP_OR_FAIL(auto task, FileScanTaskFromJson(json)); + EXPECT_EQ(task->start(), test_case.start); + EXPECT_EQ(task->length(), test_case.length); + EXPECT_EQ(task->is_split(), test_case.is_split); + ASSERT_NE(task->data_file(), nullptr); + EXPECT_EQ(*task->data_file(), DataFileForTask()); + ASSERT_EQ(task->delete_files().size(), 1U); + ASSERT_NE(task->delete_files()[0], nullptr); + EXPECT_EQ(*task->delete_files()[0], DeleteFileForTask()); + ASSERT_NE(task->residual_filter(), nullptr); + ICEBERG_UNWRAP_OR_FAIL(auto residual, ToJson(*task->residual_filter())); + EXPECT_EQ(residual, json["residual-filter"]); + + ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*task, specs_, schema_)); + EXPECT_EQ(serialized, json); + ICEBERG_UNWRAP_OR_FAIL(auto reparsed, FileScanTaskFromJson(serialized)); + EXPECT_EQ(reparsed->start(), test_case.start); + EXPECT_EQ(reparsed->length(), test_case.length); + EXPECT_EQ(reparsed->is_split(), test_case.is_split); + } +} + +TEST_F(FileScanTaskJsonTest, RejectsInvalidRanges) { + struct Case { + int64_t file_size; + int64_t start; + int64_t length; + }; + constexpr auto kMax = std::numeric_limits::max(); + for (const auto& test_case : + {Case{10, 1, 10}, // Extends beyond the file. + Case{10, 11, 1}, // Starts beyond the file. + Case{10, -1, 1}, // Negative start. + Case{10, 0, -1}, // Negative length. + Case{10, 0, 0}, // Empty range in a nonempty file. + Case{10, 10, 0}, // Empty range at EOF. + Case{0, 0, 1}, // Nonempty range in an empty file. + Case{-1, 0, 0}, // Negative file size. + Case{kMax, kMax - 1, 2}}) { // End would overflow int64_t. + SCOPED_TRACE(testing::Message() + << "file_size=" << test_case.file_size << ", start=" << test_case.start + << ", length=" << test_case.length); + auto json = JavaGolden(); + json["data-file"]["file-size-in-bytes"] = test_case.file_size; + json["start"] = test_case.start; + json["length"] = test_case.length; auto result = FileScanTaskFromJson(json); - if (test_case.supported) { - EXPECT_THAT(result, IsOk()); - } else { - EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); - EXPECT_THAT(result, HasErrorMessage("Split FileScanTask is not supported")); - } + EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(result, HasErrorMessage("Invalid FileScanTask range")); } } +TEST_F(FileScanTaskJsonTest, RoundTripsEmptyWholeFile) { + auto json = JavaGolden(); + json["data-file"]["file-size-in-bytes"] = 0; + json["data-file"]["record-count"] = 0; + json["data-file"].erase("lower-bounds"); + json["data-file"].erase("upper-bounds"); + json.erase("delete-files"); + json["start"] = 0; + json["length"] = 0; + + ICEBERG_UNWRAP_OR_FAIL(auto task, FileScanTaskFromJson(json)); + EXPECT_EQ(task->start(), 0); + EXPECT_EQ(task->length(), 0); + EXPECT_FALSE(task->is_split()); + + ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*task, specs_, schema_)); + EXPECT_EQ(serialized, json); +} + TEST_F(FileScanTaskJsonTest, RejectsFractionalIntegerFields) { for (std::string_view field : {"start", "length", "spec-id", "file-size-in-bytes", "record-count", diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc index 39427f293d..119561daf7 100644 --- a/src/iceberg/test/rest_json_serde_test.cc +++ b/src/iceberg/test/rest_json_serde_test.cc @@ -2187,7 +2187,7 @@ TEST(FileScanTasksFromJsonTest, RejectsNullOrMissingRangeEndpoint) { json[0]["length"] = nullptr; auto null_result = FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0)); EXPECT_THAT(null_result, IsError(ErrorKind::kJsonParseError)); - EXPECT_THAT(null_result, HasErrorMessage("'start' must be an integer")); + EXPECT_THAT(null_result, HasErrorMessage("Missing 'start'")); } TEST(FileScanTasksFromJsonTest, RejectsInvalidRangeIntegers) { From c097e1e3fac099e90a51094d6c3df9b63d3ef394 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Tue, 13 Oct 2026 07:03:07 +0000 Subject: [PATCH 5/8] fix(scan): report invalid standalone ranges as JSON errors --- src/iceberg/json_serde.cc | 12 +++++++----- src/iceberg/json_serde_internal.h | 9 +++++---- 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc index d64099bfd9..99b2b5e7fb 100644 --- a/src/iceberg/json_serde.cc +++ b/src/iceberg/json_serde.cc @@ -2526,11 +2526,13 @@ Result> FileScanTaskFromJson(const nlohmann::json& ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json)); } - ICEBERG_ASSIGN_OR_RAISE(auto task, FileScanTask::MakeSplit( - std::make_shared(std::move(data_file)), start, length, - std::move(delete_files), std::move(residual_filter))); - - return std::move(task); + auto task = FileScanTask::MakeSplit(std::make_shared(std::move(data_file)), + start, length, std::move(delete_files), + std::move(residual_filter)); + if (!task) { + return JsonParseError("Invalid FileScanTask range: {}", task.error().message); + } + return std::move(task.value()); } } // namespace iceberg diff --git a/src/iceberg/json_serde_internal.h b/src/iceberg/json_serde_internal.h index 1d7e06ea87..e37367826e 100644 --- a/src/iceberg/json_serde_internal.h +++ b/src/iceberg/json_serde_internal.h @@ -455,8 +455,8 @@ ICEBERG_EXPORT Result DataFileFromJson( /// - `schema` (required): schema used by this task /// - `spec` (required): partition spec used by this task /// - `data-file` (required): ContentFile JSON -/// - `start` (required): always 0 because split tasks are unsupported -/// - `length` (required): always the data file size +/// - `start` (required): inclusive byte offset where the task starts +/// - `length` (required): number of bytes in the task range /// - `delete-files` (optional on input; omitted when the task has no delete files): /// array of ContentFile JSON /// - `residual-filter` (optional): Expression JSON @@ -469,8 +469,9 @@ ICEBERG_EXPORT Result ToJson( /// \brief Deserializes a self-contained FileScanTask JSON object. /// /// Reads the embedded schema and partition spec. Rejects REST -/// `delete-file-references` and byte-range splits (`start` != 0 or `length` != file -/// size). Both range fields are required, matching Java core. +/// `delete-file-references`. Both range fields are required, matching Java core. +/// Valid ranges are preserved. C++ additionally rejects negative, out-of-file, +/// and empty partial ranges; an empty range is valid only for a zero-byte file. ICEBERG_EXPORT Result> FileScanTaskFromJson( const nlohmann::json& json); From 37912d46c34bb2e5c9ef18c51b844149639f4d03 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Tue, 13 Oct 2026 07:42:34 +0000 Subject: [PATCH 6/8] style(test): use designated initializers in RejectsInvalidRanges Co-authored-by: Isaac --- src/iceberg/test/json_serde_test.cc | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/src/iceberg/test/json_serde_test.cc b/src/iceberg/test/json_serde_test.cc index 7e95eb2ae0..efcf961515 100644 --- a/src/iceberg/test/json_serde_test.cc +++ b/src/iceberg/test/json_serde_test.cc @@ -1850,15 +1850,24 @@ TEST_F(FileScanTaskJsonTest, RejectsInvalidRanges) { }; constexpr auto kMax = std::numeric_limits::max(); for (const auto& test_case : - {Case{10, 1, 10}, // Extends beyond the file. - Case{10, 11, 1}, // Starts beyond the file. - Case{10, -1, 1}, // Negative start. - Case{10, 0, -1}, // Negative length. - Case{10, 0, 0}, // Empty range in a nonempty file. - Case{10, 10, 0}, // Empty range at EOF. - Case{0, 0, 1}, // Nonempty range in an empty file. - Case{-1, 0, 0}, // Negative file size. - Case{kMax, kMax - 1, 2}}) { // End would overflow int64_t. + {// Extends beyond the file. + Case{.file_size = 10, .start = 1, .length = 10}, + // Starts beyond the file. + Case{.file_size = 10, .start = 11, .length = 1}, + // Negative start. + Case{.file_size = 10, .start = -1, .length = 1}, + // Negative length. + Case{.file_size = 10, .start = 0, .length = -1}, + // Empty range in a nonempty file. + Case{.file_size = 10, .start = 0, .length = 0}, + // Empty range at EOF. + Case{.file_size = 10, .start = 10, .length = 0}, + // Nonempty range in an empty file. + Case{.file_size = 0, .start = 0, .length = 1}, + // Negative file size. + Case{.file_size = -1, .start = 0, .length = 0}, + // End would overflow int64_t. + Case{.file_size = kMax, .start = kMax - 1, .length = 2}}) { SCOPED_TRACE(testing::Message() << "file_size=" << test_case.file_size << ", start=" << test_case.start << ", length=" << test_case.length); From 503e3ab432e21fc0e65ac4a57643f1aeb5519219 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Tue, 13 Oct 2026 17:23:41 +0000 Subject: [PATCH 7/8] fix --- src/iceberg/catalog/rest/json_serde.cc | 1 - 1 file changed, 1 deletion(-) diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc index e6cdab23ff..54feb17263 100644 --- a/src/iceberg/catalog/rest/json_serde.cc +++ b/src/iceberg/catalog/rest/json_serde.cc @@ -18,7 +18,6 @@ */ #include -#include #include #include #include From 066fe3db0119dcd960be08b02c4a316767fc11f5 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Wed, 14 Oct 2026 20:41:44 +0000 Subject: [PATCH 8/8] fix(scan): revalidate FileScanTask range in standalone serializer A shared DataFile's size may change after MakeSplit, so ToJson could emit a range that FileScanTaskFromJson rejects. Revalidate the range before serializing, matching the REST serializer. Co-authored-by: Isaac --- src/iceberg/json_serde.cc | 7 +++++++ src/iceberg/test/json_serde_test.cc | 10 ++++++++++ 2 files changed, 17 insertions(+) diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc index 99b2b5e7fb..c604a2e0f4 100644 --- a/src/iceberg/json_serde.cc +++ b/src/iceberg/json_serde.cc @@ -2408,6 +2408,13 @@ Result ToJson( if (task.data_file()->content != DataFile::Content::kData) { return ValidationFailed("FileScanTask data-file must have data content"); } + // The shared data file may have changed size; don't write an invalid range. + auto range_result = + FileScanTask::MakeSplit(task.data_file(), task.start(), task.length()); + if (!range_result) { + return ValidationFailed("Invalid FileScanTask range: {}", + range_result.error().message); + } nlohmann::json json; json[kTaskType] = kFileScanTaskType; diff --git a/src/iceberg/test/json_serde_test.cc b/src/iceberg/test/json_serde_test.cc index efcf961515..fc7a5ec701 100644 --- a/src/iceberg/test/json_serde_test.cc +++ b/src/iceberg/test/json_serde_test.cc @@ -1882,6 +1882,16 @@ TEST_F(FileScanTaskJsonTest, RejectsInvalidRanges) { } } +TEST_F(FileScanTaskJsonTest, RejectsSplitOutsideChangedDataFile) { + auto data_file = std::make_shared(DataFileForTask()); + ICEBERG_UNWRAP_OR_FAIL(auto split, FileScanTask::MakeSplit(data_file, 4, 6)); + data_file->file_size_in_bytes = 9; + + auto result = ToJson(*split, specs_, schema_); + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT(result, HasErrorMessage("Invalid FileScanTask range")); +} + TEST_F(FileScanTaskJsonTest, RoundTripsEmptyWholeFile) { auto json = JavaGolden(); json["data-file"]["file-size-in-bytes"] = 0;