This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 26ae9eab feat(scan): Support split and parallel scans - Part-1:
validated FileScanTask byte ranges (#976)
26ae9eab is described below
commit 26ae9eab89d809a0861dfcef61ed6dc10f2b30bb
Author: kamcheungting-db <[email protected]>
AuthorDate: Wed Oct 14 20:22:33 2026 -0700
feat(scan): Support split and parallel scans - Part-1: validated
FileScanTask byte ranges (#976)
Preserve start and length in core and REST JSON, and reject partial
reads until the reader can honor ranges.
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 <[email protected]>
---
src/iceberg/catalog/rest/json_serde.cc | 42 ++++--
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/json_serde.cc | 34 ++---
src/iceberg/json_serde_internal.h | 13 +-
src/iceberg/table_scan.cc | 73 +++++++++-
src/iceberg/table_scan.h | 36 +++++
src/iceberg/test/file_scan_task_reader_test.cc | 19 +++
src/iceberg/test/file_scan_task_test.cc | 115 +++++++++++++++
src/iceberg/test/json_serde_test.cc | 105 ++++++++++++--
src/iceberg/test/rest_json_serde_test.cc | 193 +++++++++++++++++++++++--
12 files changed, 575 insertions(+), 65 deletions(-)
diff --git a/src/iceberg/catalog/rest/json_serde.cc
b/src/iceberg/catalog/rest/json_serde.cc
index a54042c2..7e90bf35 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -115,6 +115,8 @@ 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<nlohmann::json> StorageCredentialToJson(const StorageCredential&
credential) {
ICEBERG_RETURN_UNEXPECTED(credential.Validate());
@@ -187,14 +189,19 @@ Result<std::vector<std::shared_ptr<FileScanTask>>>
FileScanTasksFromJson(
ICEBERG_ASSIGN_OR_RAISE(auto data_file_json,
GetJsonValue<nlohmann::json>(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));
+ // 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) {
+ 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, GetJsonInteger<int64_t>(task_json,
kStart));
+ ICEBERG_ASSIGN_OR_RAISE(length, GetJsonInteger<int64_t>(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.
@@ -222,9 +229,13 @@ Result<std::vector<std::shared_ptr<FileScanTask>>>
FileScanTasksFromJson(
ICEBERG_ASSIGN_OR_RAISE(residual_filter,
ExpressionFromJson(filter_json));
}
- file_scan_tasks.push_back(std::make_shared<FileScanTask>(
- std::make_shared<DataFile>(std::move(data_file)),
std::move(task_delete_files),
- std::move(residual_filter)));
+ auto task =
FileScanTask::MakeSplit(std::make_shared<DataFile>(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;
}
@@ -272,10 +283,19 @@ Result<nlohmann::json> 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) {
+ 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<int32_t> refs;
diff --git a/src/iceberg/catalog/rest/types.cc
b/src/iceberg/catalog/rest/types.cc
index 156df17e..00053743 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<FileScanTask>&
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 0c41f25e..7d3a8bc7 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 a71ef5f8..c34f3e7a 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<ArrowArrayStream> Open(const FileScanTask& task);
FileScanTaskReader(const FileScanTaskReader&) = delete;
diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc
index c1c3ffb7..c604a2e0 100644
--- a/src/iceberg/json_serde.cc
+++ b/src/iceberg/json_serde.cc
@@ -2408,6 +2408,13 @@ Result<nlohmann::json> 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;
@@ -2423,8 +2430,8 @@ Result<nlohmann::json> 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 +2464,6 @@ Result<nlohmann::json> 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<std::shared_ptr<FileScanTask>> FileScanTaskFromJson(const
nlohmann::json& json) {
if (!json.is_object()) {
return JsonParseError("Cannot parse file scan task from a non-object: {}",
@@ -2510,8 +2506,6 @@ Result<std::shared_ptr<FileScanTask>>
FileScanTaskFromJson(const nlohmann::json&
ICEBERG_ASSIGN_OR_RAISE(auto start, GetJsonInteger<int64_t>(json, kStart));
ICEBERG_ASSIGN_OR_RAISE(auto length, GetJsonInteger<int64_t>(json, kLength));
- ICEBERG_RETURN_UNEXPECTED(
- CheckFileScanTaskNotSplit(start, length, data_file.file_size_in_bytes));
std::vector<std::shared_ptr<DataFile>> delete_files;
if (json.contains(kDeleteFiles)) {
@@ -2539,9 +2533,13 @@ Result<std::shared_ptr<FileScanTask>>
FileScanTaskFromJson(const nlohmann::json&
ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json));
}
- return
std::make_shared<FileScanTask>(std::make_shared<DataFile>(std::move(data_file)),
- std::move(delete_files),
- std::move(residual_filter));
+ auto task =
FileScanTask::MakeSplit(std::make_shared<DataFile>(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 2465df14..e3736782 100644
--- a/src/iceberg/json_serde_internal.h
+++ b/src/iceberg/json_serde_internal.h
@@ -455,8 +455,8 @@ ICEBERG_EXPORT Result<DataFile> 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,13 +469,10 @@ ICEBERG_EXPORT Result<nlohmann::json> 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<std::shared_ptr<FileScanTask>> 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/table_scan.cc b/src/iceberg/table_scan.cc
index 636834d6..25548bc7 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,78 @@ FileScanTask::FileScanTask(std::shared_ptr<DataFile>
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<DataFile> data_file,
+ std::vector<std::shared_ptr<DataFile>> delete_files,
+ std::shared_ptr<Expression> residual_filter, Range
range)
+ : FileScanTask(std::move(data_file), std::move(delete_files),
+ std::move(residual_filter)) {
+ range_ = range;
+}
+
+Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
+ std::shared_ptr<DataFile> data_file, int64_t start, int64_t length,
+ std::vector<std::shared_ptr<DataFile>> delete_files,
+ std::shared_ptr<Expression> 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);
+ }
+ // 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<FileScanTask>(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<FileScanTask>(
+ new FileScanTask(std::move(data_file), std::move(delete_files),
std::move(filter),
+ Range{.start = start, .length = length, .file_size =
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;
+ }
+
+ // 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<int128_t>(range_->start) + range_->length;
+ const auto count_at_end = end * record_count / range_->file_size;
+ const auto count_at_start =
+ static_cast<int128_t>(range_->start) * record_count / range_->file_size;
+ return static_cast<int64_t>(count_at_end - count_at_start);
+}
// ChangelogScanTask implementation
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index abe93bbf..7afaa79d 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 <cstdint>
#include <functional>
#include <memory>
#include <optional>
@@ -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<DataFile> data_file,
std::vector<std::shared_ptr<DataFile>> delete_files =
{},
std::shared_ptr<Expression> 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<std::shared_ptr<FileScanTask>> MakeSplit(
+ std::shared_ptr<DataFile> data_file, int64_t start, int64_t length,
+ std::vector<std::shared_ptr<DataFile>> delete_files = {},
+ std::shared_ptr<Expression> 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<DataFile>& data_file() const { return data_file_; }
@@ -92,9 +115,22 @@ 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;
+ };
+
+ FileScanTask(std::shared_ptr<DataFile> data_file,
+ std::vector<std::shared_ptr<DataFile>> delete_files,
+ std::shared_ptr<Expression> filter, Range range);
+
std::shared_ptr<DataFile> data_file_;
std::vector<std::shared_ptr<DataFile>> delete_files_;
std::shared_ptr<Expression> residual_filter_;
+ std::optional<Range> 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 cca13ce0..11366f72 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 cb945ba1..6394c96f 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 <cstdint>
+#include <limits>
+#include <memory>
+
#include <arrow/array.h>
#include <arrow/c/bridge.h>
#include <arrow/json/from_string.h>
@@ -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<DataFile>();
+ 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<DataFile>();
+ 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<DataFile>();
+ 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<DataFile>();
+ 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<int64_t>::max();
+ expect_invalid(std::numeric_limits<int64_t>::max() - 1, 2);
+}
+
+TEST(FileScanTaskRangeTest, RowEstimateDoesNotOverflow) {
+ auto data_file = std::make_shared<DataFile>();
+ data_file->file_size_in_bytes = std::numeric_limits<int64_t>::max();
+ data_file->record_count = std::numeric_limits<int64_t>::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/json_serde_test.cc
b/src/iceberg/test/json_serde_test.cc
index 51bc2af1..fc7a5ec7 100644
--- a/src/iceberg/test/json_serde_test.cc
+++ b/src/iceberg/test/json_serde_test.cc
@@ -18,6 +18,7 @@
*/
#include <array>
+#include <limits>
#include <memory>
#include <optional>
#include <string>
@@ -1803,31 +1804,113 @@ 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<int64_t>::max();
+ for (const auto& test_case :
+ {// 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);
+ 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, RejectsSplitOutsideChangedDataFile) {
+ auto data_file = std::make_shared<DataFile>(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;
+ 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 db19860a..1a593ffc 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 <limits>
#include <memory>
#include <string>
#include <string_view>
@@ -2171,6 +2172,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);
}
@@ -2190,11 +2194,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",
@@ -2205,12 +2213,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("Missing 'start'"));
}
TEST(FileScanTasksFromJsonTest, RejectsInvalidRangeIntegers) {
@@ -2246,7 +2267,7 @@ TEST(FileScanTasksFromJsonTest,
RejectsInvalidRangeIntegers) {
}
}
-TEST(FileScanTasksFromJsonTest, RejectsSplitTaskWhenEitherRangeFieldDiffers) {
+TEST(FileScanTasksFromJsonTest, AcceptsValidPartialRanges) {
auto json = R"([{
"data-file": {
"content": "data",
@@ -2261,17 +2282,85 @@ TEST(FileScanTasksFromJsonTest,
RejectsSplitTaskWhenEitherRangeFieldDiffers) {
"length": 12345
}])"_json;
- for (const auto& [field, value] :
- {std::pair<std::string_view, int64_t>{"start", 100},
- std::pair<std::string_view, int64_t>{"length", 200}}) {
- SCOPED_TRACE(field);
+ for (const auto& [start, length] :
+ {std::pair<int64_t, int64_t>{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<int64_t, int64_t>{-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<int64_t>::max();
+ overflow_json[0]["start"] = std::numeric_limits<int64_t>::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) {
@@ -2444,6 +2533,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<DataFile>();
+ 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<std::shared_ptr<FileScanTask>>{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": [
