You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/01 11:58:24 UTC
(iggy) branch fix/iceberg-partition-routing updated (4aaa0d313 -> 3593e95ee)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch fix/iceberg-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit 4aaa0d313 fix(connectors): write iceberg sink data into real table partitions
add 1e2d2220f fix(sdk): error callback should await shard flush on producer shutdown (#3953)
add 5068a960a chore(deps): Bump the github-actions group across 1 directory with 2 updates (#3968)
add d7c2b6180 fix(python): read consumer metadata without the consumer lock (#3888)
add 03f5b3034 fix(server): restore the mimalloc global allocator (#3971)
add ed59b955d chore(deps): Bump gradle from 9.6.1-jdk17 to 9.7.0-jdk17 in /bdd/java in the docker group across 1 directory (#3972)
add 37aec0877 feat(sdk): fail over to a surviving node when the current one dies (#3944)
add 6a1c84c39 fix(partitions): stop reserving 16 MiB per active segment index (#3974)
add ffb186dba feat(gateways): iggy gateway for kafka listener (#3519)
add 9c6451d62 fix(consensus): keep a late retry's reply instead of refusing it (#3945)
add 7b491c09f fix(cluster): prevent repaired prepares from rewinding WAL parent (#3978)
add 76f2dda5d feat(simulator): drive the workload under network and replica faults (#3942)
add 1ad3f654c chore(deps): Bump the web group across 1 directory with 4 updates (#3983)
add 0fdf9d688 chore(deps): Bump the node group across 1 directory with 10 updates (#3984)
add db14a4b17 fix(cluster): make the advertised leader accept partition writes (#3985)
add 90e38b10f feat(node): support connection string syntax for nodejs client creation (#3917)
add 996ac04c8 fix(cluster): recover writes after a fast partition primary rejoin (#3987)
add 5916e6f58 feat(partitions): persist log and index concurrently, harden recovery (#3970)
add eb410a96a fix(server): apply thread pool limit carve-out to all of macOS (#3994)
add c8ef3b9a4 fix(go): tear down only the connection a request failed on (#3986)
add 8b7dbd9c6 chore(deps): Bump the go group across 2 directories with 2 updates (#3990)
add a1df38354 test(partitions): gate the persist-retry test on Linux (#3998)
add 31b86ff2a fix(server)!: reject a wildcard bind with no advertised address (#3923)
add dde679aea ci(python): fold the build task into test and check the stub (#3980)
add 3e4898ac6 fix(bdd): resolve server address and credentials only from env (#3935)
add e80992b63 chore(deps): Bump the java group across 3 directories with 17 updates (#3966)
add dc1cdf232 fix(sdk): respect max_buffer_size when merging batches (#3933)
add 6ef0c1677 feat(sdk): consume the request counter for partition operations (#3958)
add e9b1c9a12 fix(cluster): stop a late partition materialiser from wedging its group (#4007)
add bb973973d chore(deps): Bump the csharp group with 1 update (#4011)
add b1b911ffb fix(sdk): stream should terminate after consumer shutdown (#3964)
add 72b5995c5 docs(repo): add repo-wide team-review agent skill (#4004)
add 328b28974 fix(journal): reject recovery slot collisions (#4009)
add 8c4986ac5 fix(io_uring): keep shard startup off blocking fallbacks
new 14649bae0 fix(connectors): write iceberg sink data into real table partitions
new ddcf75a74 chore(deps): unify arrow/parquet on 58 and bump iceberg to 0.10.1
new 3593e95ee chore(connectors): mark clickhouse sink as unpublishable
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (4aaa0d313)
\
N -- N -- N refs/heads/fix/iceberg-partition-routing (3593e95ee)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.claude/skills/team-review/SKILL.md | 117 +
.devcontainer/devcontainer.json | 3 +-
.github/actions/go/pre-merge/action.yml | 2 +
.github/actions/php/pre-merge/action.yml | 10 +-
.../actions/python-maturin/pre-merge/action.yml | 50 +-
.github/actions/rust/pre-merge/action.yml | 38 +
.github/config/components.yml | 12 +-
.github/workflows/_common.yml | 4 +-
.github/workflows/_test_examples.yml | 2 +-
.github/workflows/coverage-baseline.yml | 10 +-
.github/workflows/post-merge.yml | 2 +-
.github/workflows/pr-title.yml | 1 +
.gitignore | 1 +
AGENTS.md | 17 +-
Cargo.lock | 1001 +++--
Cargo.toml | 24 +-
README.md | 2 +-
.../features/step_definitions/background_steps.cpp | 15 +-
bdd/docker-compose.server.yml | 4 +
bdd/docker-compose.yml | 9 +-
bdd/go/go.mod | 4 +-
bdd/go/go.sum | 19 +-
bdd/go/tests/basic_messaging.go | 10 +-
bdd/go/tests/env/env.go | 55 +
bdd/go/tests/leader_redirection.go | 26 +-
bdd/go/tests/raw_command.go | 11 +-
bdd/go/tests/tcp_test/session_feature_login.go | 7 +-
bdd/go/tests/tcp_test/test_helpers.go | 11 +-
bdd/java/Dockerfile | 2 +-
bdd/java/build.gradle.kts | 10 +-
bdd/java/gradle/wrapper/gradle-wrapper.properties | 2 +-
bdd/java/gradlew | 4 +-
.../org/apache/iggy/bdd/BasicMessagingSteps.java | 11 +-
.../apache/iggy/bdd/LeaderRedirectionSteps.java | 15 +-
.../java/org/apache/iggy/bdd/TestEnvironment.java | 59 +
bdd/php/Dockerfile | 2 -
bdd/php/tests/BasicMessagingFeatureTest.php | 5 +-
bdd/php/tests/RawCommandFeatureTest.php | 7 +-
bdd/php/tests/TestEnvironment.php | 53 +
bdd/python/tests/conftest.py | 24 +-
bdd/python/tests/test_basic_messaging.py | 4 +-
bdd/python/tests/test_raw_command.py | 4 +-
bdd/python/uv.lock | 2 +-
bdd/rust/tests/helpers/cluster.rs | 36 +-
.../rust/tests/helpers/env.rs | 57 +-
bdd/rust/tests/helpers/mod.rs | 1 +
bdd/rust/tests/steps/auth.rs | 3 +-
bdd/rust/tests/steps/leader_redirection.rs | 17 +-
bdd/rust/tests/steps/server.rs | 5 +-
bdd/rust/tests/steps/streams.rs | 22 +-
core/ai/mcp/Cargo.toml | 2 +-
core/bench/Cargo.toml | 2 +-
core/bench/report/src/types/group_metrics.rs | 28 +-
core/bench/report/src/types/individual_metrics.rs | 94 +-
core/bench/report/src/utils.rs | 30 +-
.../actors/consumer/typed_benchmark_consumer.rs | 12 +-
core/bench/src/analytics/metrics/group.rs | 48 +-
core/bench/src/analytics/metrics/individual.rs | 65 +-
core/bench/src/args/common.rs | 17 +-
core/bench/src/args/examples.rs | 6 +
core/bench/src/args/kinds/balanced/producer.rs | 19 +-
core/bench/src/benchmarks/benchmark.rs | 13 +-
core/binary_protocol/Cargo.toml | 2 +-
core/binary_protocol/src/consensus/header.rs | 4 +-
core/binary_protocol/src/lib.rs | 2 +-
core/binary_protocol/src/primitives/consumer.rs | 3 +-
.../create_partitions_with_assignments.rs | 44 +
.../topics/create_topic_with_assignments.rs | 46 +
core/cli/Cargo.toml | 2 +-
core/common/Cargo.toml | 2 +-
.../traits/binary_impls/personal_access_tokens.rs | 10 +-
core/common/src/traits/binary_impls/users.rs | 14 +-
core/common/src/traits/binary_transport.rs | 24 +-
.../tcp_config/tcp_client_reconnection_config.rs | 35 +-
.../configs/src/configs_impl/typed_env_provider.rs | 13 +-
core/configs/src/server_config/cluster.rs | 423 +-
core/configs/src/server_config/defaults.rs | 2 +
core/configs/src/server_config/metadata.rs | 4 +-
core/configs/src/server_config/mod.rs | 1 +
core/configs/src/server_config/node.rs | 89 +
core/configs/src/server_config/server.rs | 54 +
core/configs/src/server_config/validators.rs | 133 +
core/connectors/runtime/Cargo.toml | 2 +-
core/connectors/sinks/clickhouse_sink/Cargo.toml | 1 +
core/connectors/sinks/iceberg_sink/src/catalog.rs | 1 -
core/connectors/sinks/iceberg_sink/src/props.rs | 4 +
.../sinks/iceberg_sink/src/router/mod.rs | 26 +-
core/consensus/src/client_table.rs | 290 +-
core/consensus/src/impls.rs | 115 +-
core/consensus/src/metadata_helpers.rs | 8 +-
core/consensus/src/plane_helpers.rs | 62 +
core/harness_derive/src/attrs.rs | 12 +
core/harness_derive/src/codegen.rs | 6 +
core/integration/src/harness/disk.rs | 19 +-
.../src/harness/handle/client_builder.rs | 61 +-
.../src/harness/orchestrator/harness.rs | 145 +-
.../tests/cluster/client_table_adversarial.rs | 194 +-
.../tests/cluster/crash_recovery_corruption.rs | 269 ++
.../tests/cluster/failover_client_continuity.rs | 47 +-
.../tests/cluster/fast_primary_rejoin.rs | 472 ++
core/integration/tests/cluster/mod.rs | 3 +
.../tests/cluster/partition_primary_routing.rs | 198 +
.../tests/cluster/staggered_bootstrap.rs | 180 +
core/integration/tests/sdk/disconnect_relogin.rs | 66 +
core/integration/tests/sdk/mod.rs | 1 +
.../tests/server/cluster_metadata_advertised.rs | 80 +
core/integration/tests/server/mod.rs | 5 +-
core/journal/src/file_storage.rs | 78 +-
core/journal/src/prepare_journal.rs | 135 +-
core/message_bus/src/client_listener/mod.rs | 75 +-
core/message_bus/src/client_listener/tcp.rs | 8 +-
core/message_bus/src/client_listener/ws.rs | 8 +-
core/message_bus/src/replica/io.rs | 4 +-
core/message_bus/tests/graceful_shutdown.rs | 4 +-
core/message_bus/tests/tcp_client_roundtrip.rs | 4 +-
core/message_bus/tests/ws_client_roundtrip.rs | 4 +-
core/metadata/src/impls/metadata.rs | 254 +-
core/metadata/src/impls/mod.rs | 5 +
core/metadata/src/stm/authz.rs | 1 +
core/metadata/src/stm/consumer_group.rs | 1 +
core/metadata/src/stm/snapshot.rs | 33 +-
core/metadata/src/stm/stream.rs | 126 +-
core/partitions/src/iggy_index.rs | 6 +
core/partitions/src/iggy_index_reader.rs | 3 +-
core/partitions/src/iggy_index_writer.rs | 24 +-
core/partitions/src/iggy_partition.rs | 1102 ++++-
core/partitions/src/journal.rs | 31 +
core/partitions/src/lib.rs | 30 +-
core/partitions/src/log.rs | 81 +-
core/partitions/src/messages_writer.rs | 35 +-
core/partitions/src/poll_plan.rs | 83 +-
core/partitions/src/state_transfer.rs | 8 +
core/partitions/src/types.rs | 35 +-
core/sdk/Cargo.toml | 2 +-
.../src/clients/binary_personal_access_tokens.rs | 10 +
core/sdk/src/clients/binary_users.rs | 10 +
core/sdk/src/clients/consumer.rs | 456 +-
core/sdk/src/clients/mod.rs | 33 +
core/sdk/src/clients/producer_config.rs | 3 +-
core/sdk/src/clients/producer_dispatcher.rs | 256 +-
core/sdk/src/clients/producer_sharding.rs | 282 +-
core/sdk/src/leader_aware.rs | 835 +++-
core/sdk/src/prelude.rs | 2 +-
core/sdk/src/quic/quic_client.rs | 375 +-
core/sdk/src/tcp/tcp_client.rs | 2011 ++++++++-
core/sdk/src/vsr.rs | 111 +-
core/sdk/src/websocket/websocket_client.rs | 424 +-
core/server/Cargo.toml | 2 +-
core/server/README.md | 11 +-
core/server/config.toml | 99 +-
core/server/src/args.rs | 4 +-
core/server/src/bootstrap.rs | 396 +-
core/server/src/cluster_meta.rs | 196 +-
core/server/src/dispatch.rs | 24 +-
core/server/src/http.rs | 52 +-
core/server/src/http/error.rs | 19 +-
core/server/src/http/forward.rs | 203 +-
core/server/src/http/session.rs | 5 +-
core/server/src/lib.rs | 8 +
core/server/src/main.rs | 4 +
core/server/src/partition_helpers.rs | 46 +-
core/server/src/partition_reconciler.rs | 61 +-
core/server/src/responses.rs | 8 +-
core/server/src/segment_recovery.rs | 2566 +++++++++--
core/server/src/server_error.rs | 133 +-
core/server_common/src/executor.rs | 2 +-
core/shard/Cargo.toml | 4 +-
core/shard/src/lib.rs | 354 +-
core/shard/src/router.rs | 110 +-
core/simulator/Cargo.toml | 4 +
core/simulator/src/bin/workload-fuzz.rs | 789 +++-
core/simulator/src/client.rs | 185 +-
core/simulator/src/deps.rs | 159 +-
core/simulator/src/executor/mod.rs | 12 +-
core/simulator/src/lib.rs | 4752 +++++++++++++-------
core/simulator/src/network.rs | 23 +-
core/simulator/src/packet.rs | 368 +-
core/simulator/src/ready_queue.rs | 4 +-
core/simulator/src/replica.rs | 123 +-
core/simulator/src/seeds.rs | 111 +
core/simulator/src/workload/auditor.rs | 66 +-
core/simulator/src/workload/ids.rs | 4 +-
core/simulator/src/workload/invariants.rs | 27 +-
core/simulator/src/workload/mod.rs | 864 +++-
core/simulator/src/workload/ops/change_password.rs | 4 +-
.../src/workload/ops/create_consumer_group.rs | 4 +-
.../src/workload/ops/create_partitions.rs | 4 +-
.../workload/ops/create_personal_access_token.rs | 4 +-
core/simulator/src/workload/ops/create_stream.rs | 4 +-
core/simulator/src/workload/ops/create_topic.rs | 4 +-
core/simulator/src/workload/ops/create_user.rs | 4 +-
.../src/workload/ops/delete_consumer_group.rs | 4 +-
.../src/workload/ops/delete_consumer_offset.rs | 7 +-
.../src/workload/ops/delete_partitions.rs | 4 +-
.../workload/ops/delete_personal_access_token.rs | 4 +-
core/simulator/src/workload/ops/delete_segments.rs | 4 +-
core/simulator/src/workload/ops/delete_stream.rs | 4 +-
core/simulator/src/workload/ops/delete_topic.rs | 4 +-
core/simulator/src/workload/ops/delete_user.rs | 4 +-
core/simulator/src/workload/ops/mod.rs | 24 +-
core/simulator/src/workload/ops/purge_stream.rs | 4 +-
core/simulator/src/workload/ops/purge_topic.rs | 4 +-
core/simulator/src/workload/ops/send_messages.rs | 4 +-
.../src/workload/ops/store_consumer_offset.rs | 7 +-
.../src/workload/ops/update_permissions.rs | 4 +-
core/simulator/src/workload/ops/update_stream.rs | 4 +-
core/simulator/src/workload/ops/update_topic.rs | 4 +-
core/simulator/src/workload/ops/update_user.rs | 4 +-
core/simulator/src/workload/options.rs | 135 +-
core/simulator/src/workload/oracle.rs | 505 ++-
core/simulator/src/workload/shadow.rs | 14 +-
core/simulator/src/workload/state_checker.rs | 434 ++
docker-compose.yml | 4 +
examples/csharp/README.md | 2 +-
examples/go/README.md | 2 +-
examples/go/go.sum | 10 +-
examples/java/README.md | 2 +-
examples/java/build.gradle.kts | 4 +-
.../java/gradle/wrapper/gradle-wrapper.properties | 2 +-
examples/java/gradlew | 4 +-
examples/node/README.md | 5 +-
examples/node/package-lock.json | 2 +-
examples/python/README.md | 5 +-
examples/python/uv.lock | 2 +-
examples/rust/README.md | 2 +-
foreign/cpp/Cargo.toml | 2 +-
foreign/cpp/MODULE.bazel | 2 +-
foreign/csharp/Directory.Packages.props | 2 +-
.../Iggy_SDK.Tests.BDD/Context/TestContext.cs | 6 +-
.../Iggy_SDK.Tests.BDD/Context/TestEnvironment.cs | 44 +
.../csharp/Iggy_SDK.Tests.BDD/Context/TestHooks.cs | 3 -
foreign/csharp/Iggy_SDK.Tests.BDD/README.md | 25 +-
.../BasicMessagingOperationsSteps.cs | 3 +-
.../StepDefinitions/LeaderRedirectionSteps.cs | 6 +-
.../Fixtures/VsrCluster.cs | 1 +
.../Iggy_SDK.Tests.Integration/HeartbeatTests.cs | 31 +-
.../Implementations/TcpMessageStream.Vsr.cs | 116 +-
.../IggyClient/Implementations/TcpMessageStream.cs | 246 +-
foreign/csharp/Iggy_SDK/Iggy_SDK.csproj | 2 +-
foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs | 10 +-
foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs | 6 +-
foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs | 11 -
.../VsrTests/ConsensusSessionTests.cs | 19 +-
.../Iggy_SDK_Tests/VsrTests/DialCandidatesTests.cs | 79 +
.../VsrTests/EndpointFailoverTests.cs | 702 +++
.../Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs | 12 +-
.../Iggy_SDK_Tests/VsrTests/VsrOperationTests.cs | 5 +-
foreign/csharp/README.md | 27 +-
foreign/go/client/tcp/tcp_connect_test.go | 15 +-
foreign/go/client/tcp/tcp_core.go | 551 ++-
foreign/go/client/tcp/tcp_core_review_test.go | 13 +-
foreign/go/client/tcp/tcp_failover_test.go | 846 ++++
.../go/client/tcp/tcp_session_credentials_test.go | 66 +
foreign/go/client/tcp/tcp_session_management.go | 182 +-
foreign/go/contracts/version.go | 2 +-
foreign/go/go.mod | 5 +-
foreign/go/go.sum | 10 +-
foreign/go/internal/vsr/envelope.go | 11 +-
foreign/go/internal/vsr/envelope_test.go | 18 +-
foreign/go/internal/vsr/header.go | 5 +-
foreign/go/internal/vsr/operation.go | 11 +-
foreign/go/internal/vsr/operation_test.go | 8 -
foreign/go/internal/vsr/protocol_parity_test.go | 3 -
foreign/go/internal/vsr/session.go | 11 +-
.../iggy-connector-flink/docker-compose.yml | 1 +
.../iggy-connector-pinot/docker-compose.yml | 1 +
foreign/java/gradle/libs.versions.toml | 16 +-
.../java/gradle/wrapper/gradle-wrapper.properties | 2 +-
foreign/java/gradlew | 4 +-
.../iggy/client/async/tcp/AsyncIggyTcpClient.java | 408 +-
.../iggy/client/async/tcp/AsyncTcpConnection.java | 116 +-
.../iggy/client/async/tcp/LeaderAwareness.java | 109 +-
.../iggy/client/async/tcp/LoginRoutingHook.java | 6 +
.../iggy/client/async/tcp/ReconnectPlan.java | 17 +-
.../iggy/client/async/tcp/UsersTcpClient.java | 1 +
.../client/async/tcp/vsr/ConsensusSession.java | 30 +-
.../iggy/client/async/tcp/vsr/VsrHeaders.java | 6 +-
.../iggy/client/async/tcp/vsr/VsrOperation.java | 5 -
.../client/async/tcp/vsr/VsrRequestEncoder.java | 9 +-
.../client/async/tcp/vsr/VsrResponseHandler.java | 11 +-
.../apache/iggy/client/BaseIntegrationTest.java | 1 +
.../AsyncIggyTcpClientEndpointFailoverTest.java | 562 +++
.../AsyncIggyTcpClientTransientFailoverTest.java | 323 +-
.../tcp/AsyncTcpConnectionConcurrencyTest.java | 4 +-
.../iggy/client/async/tcp/LeaderAwarenessTest.java | 141 +-
.../iggy/client/async/tcp/ReconnectPlanTest.java | 18 -
.../async/tcp/vsr/VsrRequestEncoderTest.java | 12 +-
.../async/tcp/vsr/VsrResponseHandlerTest.java | 82 +-
foreign/node/README.md | 59 +-
foreign/node/package-lock.json | 247 +-
foreign/node/package.json | 2 +-
foreign/node/scripts/check-vsr-protocol.mjs | 6 -
foreign/node/src/bdd/README.md | 23 +-
foreign/node/src/bdd/auth.ts | 6 +-
.../types/mod.rs => foreign/node/src/bdd/env.ts | 38 +-
foreign/node/src/client/client.config.test.ts | 44 +
foreign/node/src/client/client.config.ts | 33 +-
.../src/client/client.connection-string.test.ts | 310 ++
.../node/src/client/client.connection-string.ts | 226 +
foreign/node/src/client/client.connection.test.ts | 503 ++-
foreign/node/src/client/client.connection.ts | 333 +-
foreign/node/src/client/client.socket.test.ts | 624 ++-
foreign/node/src/client/client.socket.ts | 439 +-
foreign/node/src/client/client.ts | 57 +-
foreign/node/src/client/client.type.ts | 51 +-
foreign/node/src/client/index.ts | 1 +
.../theme/mod.rs => foreign/node/src/constant.ts | 3 +-
foreign/node/src/duration.utils.test.ts | 180 +
foreign/node/src/duration.utils.ts | 393 ++
foreign/node/src/e2e/tcp.connection-string.e2e.ts | 146 +
foreign/node/src/e2e/tls.system.e2e.ts | 64 +
foreign/node/src/stream/consumer-stream.ts | 6 +-
foreign/node/src/wire/vsr/header.ts | 6 +-
foreign/node/src/wire/vsr/index.ts | 13 +-
foreign/node/src/wire/vsr/operation.test.ts | 3 -
foreign/node/src/wire/vsr/operation.ts | 5 -
foreign/node/src/wire/vsr/session.ts | 6 +-
foreign/node/src/wire/vsr/vsr.test.ts | 43 +-
foreign/php/Cargo.toml | 2 +-
foreign/php/Dockerfile.test | 3 +-
foreign/php/README.md | 3 +-
foreign/php/docker-compose.test.yml | 8 +-
foreign/php/scripts/test.sh | 9 +-
foreign/php/tests/IggySdkTest.php | 2 +-
foreign/php/tests/bootstrap.php | 17 +-
foreign/python/Cargo.toml | 4 +-
foreign/python/README.md | 3 +-
foreign/python/apache_iggy.pyi | 81 +-
foreign/python/docker-compose.test.yml | 1 +
foreign/python/pyproject.toml | 6 +-
foreign/python/src/bin/stub_gen.rs | 11 +-
foreign/python/src/client.rs | 8 +
foreign/python/src/config.rs | 19 +-
foreign/python/src/consumer.rs | 122 +-
foreign/python/src/topic.rs | 4 +-
foreign/python/tests/test_consumer_group.py | 118 +
foreign/python/tests/test_tls.py | 1 +
foreign/python/uv.lock | 2 +-
gateways/README.md | 9 +
{core/bench => gateways/kafka}/Cargo.toml | 57 +-
gateways/kafka/README.md | 63 +
gateways/kafka/docs/MANUAL_TESTING.md | 307 ++
gateways/kafka/docs/SCOPE.md | 138 +
gateways/kafka/docs/TEST_SUITE.md | 66 +
gateways/kafka/docs/kafka_api_keys_reference.md | 316 ++
gateways/kafka/scripts/ci-wire-fixtures.sh | 57 +
gateways/kafka/src/error.rs | 45 +
.../steps/mod.rs => gateways/kafka/src/lib.rs | 12 +-
gateways/kafka/src/main.rs | 240 +
gateways/kafka/src/protocol/api.rs | 637 +++
gateways/kafka/src/protocol/bounds_guard.rs | 906 ++++
gateways/kafka/src/protocol/header.rs | 40 +
.../helpers => gateways/kafka/src/protocol}/mod.rs | 6 +-
gateways/kafka/src/protocol/responses.rs | 318 ++
gateways/kafka/src/server.rs | 896 ++++
gateways/kafka/tests/api_handler_tests.rs | 989 ++++
gateways/kafka/tests/broker_advertise_tests.rs | 117 +
gateways/kafka/tests/common/codec.rs | 292 ++
gateways/kafka/tests/common/fixtures.rs | 91 +
.../kafka/tests/common/scope.rs | 26 +-
gateways/kafka/tests/common/server.rs | 68 +
gateways/kafka/tests/common/tcp.rs | 258 ++
gateways/kafka/tests/common/wire.rs | 447 ++
.../kafka/tests/fixtures_canary_tests.rs | 36 +-
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 122 +
gateways/kafka/tests/header_tests.rs | 282 ++
gateways/kafka/tests/listener_robustness_tests.rs | 624 +++
gateways/kafka/tests/response_negative_tests.rs | 277 ++
gateways/kafka/tests/server_e2e_tests.rs | 451 ++
gateways/kafka/tests/server_integration_tests.rs | 129 +
gateways/kafka/tests/version_firewall_tests.rs | 799 ++++
.../kafka/tools/kafka-tool}/Cargo.toml | 20 +-
gateways/kafka/tools/kafka-tool/README.md | 251 ++
gateways/kafka/tools/kafka-tool/src/main.rs | 864 ++++
gateways/kafka/tools/kafka-tool/src/response.rs | 389 ++
.../tools/kafka-tool/tests/generate_cli_tests.rs | 82 +
helm/charts/iggy/README.md | 31 +-
helm/charts/iggy/README.md.gotmpl | 28 +-
helm/charts/iggy/templates/deployment.yaml | 16 +
helm/charts/iggy/values.yaml | 5 +
scripts/ci/coverage-baseline-affected.sh | 2 +-
web/README.md | 3 +-
web/docker-compose.yml | 1 +
web/package-lock.json | 182 +-
web/package.json | 8 +-
385 files changed, 39680 insertions(+), 5983 deletions(-)
create mode 100644 .claude/skills/team-review/SKILL.md
create mode 100644 bdd/go/tests/env/env.go
create mode 100644 bdd/java/src/test/java/org/apache/iggy/bdd/TestEnvironment.java
create mode 100644 bdd/php/tests/TestEnvironment.php
copy foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs => bdd/rust/tests/helpers/env.rs (50%)
create mode 100644 core/configs/src/server_config/node.rs
create mode 100644 core/integration/tests/cluster/fast_primary_rejoin.rs
create mode 100644 core/integration/tests/cluster/partition_primary_routing.rs
create mode 100644 core/integration/tests/cluster/staggered_bootstrap.rs
create mode 100644 core/integration/tests/sdk/disconnect_relogin.rs
create mode 100644 core/integration/tests/server/cluster_metadata_advertised.rs
create mode 100644 core/simulator/src/seeds.rs
create mode 100644 core/simulator/src/workload/state_checker.rs
create mode 100644 foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestEnvironment.cs
create mode 100644 foreign/csharp/Iggy_SDK_Tests/VsrTests/DialCandidatesTests.cs
create mode 100644 foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
create mode 100644 foreign/go/client/tcp/tcp_failover_test.go
create mode 100644 foreign/go/client/tcp/tcp_session_credentials_test.go
create mode 100644 foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEndpointFailoverTest.java
copy core/bench/report/src/types/mod.rs => foreign/node/src/bdd/env.ts (50%)
create mode 100644 foreign/node/src/client/client.connection-string.test.ts
create mode 100644 foreign/node/src/client/client.connection-string.ts
copy core/bench/dashboard/frontend/src/components/theme/mod.rs => foreign/node/src/constant.ts (89%)
create mode 100644 foreign/node/src/duration.utils.test.ts
create mode 100644 foreign/node/src/duration.utils.ts
create mode 100644 foreign/node/src/e2e/tcp.connection-string.e2e.ts
create mode 100644 gateways/README.md
copy {core/bench => gateways/kafka}/Cargo.toml (54%)
create mode 100644 gateways/kafka/README.md
create mode 100644 gateways/kafka/docs/MANUAL_TESTING.md
create mode 100644 gateways/kafka/docs/SCOPE.md
create mode 100644 gateways/kafka/docs/TEST_SUITE.md
create mode 100644 gateways/kafka/docs/kafka_api_keys_reference.md
create mode 100755 gateways/kafka/scripts/ci-wire-fixtures.sh
create mode 100644 gateways/kafka/src/error.rs
copy bdd/rust/tests/steps/mod.rs => gateways/kafka/src/lib.rs (85%)
create mode 100644 gateways/kafka/src/main.rs
create mode 100644 gateways/kafka/src/protocol/api.rs
create mode 100644 gateways/kafka/src/protocol/bounds_guard.rs
create mode 100644 gateways/kafka/src/protocol/header.rs
copy {bdd/rust/tests/helpers => gateways/kafka/src/protocol}/mod.rs (91%)
create mode 100644 gateways/kafka/src/protocol/responses.rs
create mode 100644 gateways/kafka/src/server.rs
create mode 100644 gateways/kafka/tests/api_handler_tests.rs
create mode 100644 gateways/kafka/tests/broker_advertise_tests.rs
create mode 100644 gateways/kafka/tests/common/codec.rs
create mode 100644 gateways/kafka/tests/common/fixtures.rs
copy bdd/rust/tests/stream_crud.rs => gateways/kafka/tests/common/scope.rs (60%)
create mode 100644 gateways/kafka/tests/common/server.rs
create mode 100644 gateways/kafka/tests/common/tcp.rs
create mode 100644 gateways/kafka/tests/common/wire.rs
copy foreign/csharp/Iggy_SDK_Tests/Utils/DummyObj/DummyObjFactory.cs => gateways/kafka/tests/fixtures_canary_tests.rs (51%)
create mode 100644 gateways/kafka/tests/golden_wire_fixtures_tests.rs
create mode 100644 gateways/kafka/tests/header_tests.rs
create mode 100644 gateways/kafka/tests/listener_robustness_tests.rs
create mode 100644 gateways/kafka/tests/response_negative_tests.rs
create mode 100644 gateways/kafka/tests/server_e2e_tests.rs
create mode 100644 gateways/kafka/tests/server_integration_tests.rs
create mode 100644 gateways/kafka/tests/version_firewall_tests.rs
copy {core/bench/runner => gateways/kafka/tools/kafka-tool}/Cargo.toml (73%)
create mode 100644 gateways/kafka/tools/kafka-tool/README.md
create mode 100644 gateways/kafka/tools/kafka-tool/src/main.rs
create mode 100644 gateways/kafka/tools/kafka-tool/src/response.rs
create mode 100644 gateways/kafka/tools/kafka-tool/tests/generate_cli_tests.rs
(iggy) 02/03: chore(deps): unify arrow/parquet on 58 and bump iceberg to 0.10.1
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/iceberg-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit ddcf75a7460cd5e7d5633d8d673c52fadaeae15c
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Sep 1 13:51:55 2026 +0200
chore(deps): unify arrow/parquet on 58 and bump iceberg to 0.10.1
The workspace pinned arrow/parquet 57 while deltalake pulled 58, so
the lock carried two copies of the whole arrow family (16 crates)
plus parquet. iceberg 0.9 was the last consumer holding us on 57.
Move to arrow/parquet 58.4 and iceberg 0.10.1 so every consumer
shares one arrow. iceberg 0.10 drops configured_scheme from the S3
storage factory, borrows table metadata in DefaultLocationGenerator,
requires an explicit Runtime when building a Table, and switches the
S3 default to virtual-host-style addressing, so the sink now pins
s3.path-style-access=true to keep MinIO-style endpoints working.
thrift 0.17 (GHSA-2f9f-gq7v-9h6m) stays until parquet 59 lands in
iceberg-rust and deltalake; parquet only uses its output protocol.
---
Cargo.lock | 919 ++++++++++-----------
Cargo.toml | 14 +-
core/connectors/sinks/iceberg_sink/src/catalog.rs | 1 -
core/connectors/sinks/iceberg_sink/src/props.rs | 4 +
.../sinks/iceberg_sink/src/router/mod.rs | 26 +-
5 files changed, 471 insertions(+), 493 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 1eb6e89f3..aa6f893c7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -67,7 +67,7 @@ dependencies = [
"actix-rt",
"actix-service",
"actix-utils",
- "base64",
+ "base64 0.22.1",
"bitflags 2.13.1",
"brotli",
"bytes",
@@ -229,6 +229,16 @@ version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
+[[package]]
+name = "aead"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0"
+dependencies = [
+ "crypto-common 0.1.7",
+ "generic-array",
+]
+
[[package]]
name = "aead"
version = "0.6.1"
@@ -261,17 +271,31 @@ dependencies = [
"cpufeatures 0.3.0",
]
+[[package]]
+name = "aes-gcm"
+version = "0.10.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1"
+dependencies = [
+ "aead 0.5.2",
+ "aes 0.8.4",
+ "cipher 0.4.4",
+ "ctr 0.9.2",
+ "ghash 0.5.1",
+ "subtle",
+]
+
[[package]]
name = "aes-gcm"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fdf011db2e21ce0d575593d749db5554b47fed37aff429e4dc50bc91ac93a028"
dependencies = [
- "aead",
+ "aead 0.6.1",
"aes 0.9.1",
"cipher 0.5.2",
- "ctr",
- "ghash",
+ "ctr 0.10.1",
+ "ghash 0.6.0",
"subtle",
]
@@ -422,6 +446,7 @@ checksum = "36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf"
dependencies = [
"bigdecimal",
"bon",
+ "crc32fast",
"digest 0.10.7",
"log",
"miniz_oxide",
@@ -432,6 +457,7 @@ dependencies = [
"serde",
"serde_bytes",
"serde_json",
+ "snap",
"strum 0.27.2",
"strum_macros 0.27.2",
"thiserror 2.0.19",
@@ -517,102 +543,49 @@ checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
[[package]]
name = "arrow"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3bd47f2a6ddc39244bd722a27ee5da66c03369d087b9e024eafdb03e98b98ea7"
-dependencies = [
- "arrow-arith 57.3.1",
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-cast 57.3.1",
- "arrow-csv 57.3.1",
- "arrow-data 57.3.1",
- "arrow-ipc 57.3.1",
- "arrow-json 57.3.1",
- "arrow-ord 57.3.1",
- "arrow-row 57.3.1",
- "arrow-schema 57.3.1",
- "arrow-select 57.3.1",
- "arrow-string 57.3.1",
-]
-
-[[package]]
-name = "arrow"
-version = "58.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "378530e55cd479eda3c14eb345310799717e6f76d0c332041e8487022166b471"
-dependencies = [
- "arrow-arith 58.3.0",
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-cast 58.3.0",
- "arrow-csv 58.3.0",
- "arrow-data 58.3.0",
- "arrow-ipc 58.3.0",
- "arrow-json 58.3.0",
- "arrow-ord 58.3.0",
- "arrow-row 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
- "arrow-string 58.3.0",
-]
-
-[[package]]
-name = "arrow-arith"
-version = "57.3.1"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c7bbd679c5418b8639b92be01f361d60013c4906574b578b77b63c78356594c"
+checksum = "6cfdd0833e32a9874d2b55089333ad310c0be208aafa277385ce2461dec90be3"
dependencies = [
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
- "chrono",
- "num-traits",
+ "arrow-arith",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-cast",
+ "arrow-csv",
+ "arrow-data",
+ "arrow-ipc",
+ "arrow-json",
+ "arrow-ord",
+ "arrow-row",
+ "arrow-schema",
+ "arrow-select",
+ "arrow-string",
]
[[package]]
name = "arrow-arith"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a0ab212d2c1886e802f51c5212d78ebbcbb0bec980fff9dadc1eb8d45cd0b738"
+checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a"
dependencies = [
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-schema 58.3.0",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
"chrono",
"num-traits",
]
[[package]]
name = "arrow-array"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8a4ab47b3f3eac60f7fd31b81e9028fda018607bcc63451aca4f2b755269862"
-dependencies = [
- "ahash 0.8.12",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
- "chrono",
- "half",
- "hashbrown 0.16.1",
- "num-complex",
- "num-integer",
- "num-traits",
-]
-
-[[package]]
-name = "arrow-array"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cfd33d3e92f207444098c75b42de99d329562be0cf686b307b097cc52b4e999e"
+checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69"
dependencies = [
"ahash 0.8.12",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-schema 58.3.0",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
"chrono",
"chrono-tz",
"half",
@@ -624,21 +597,9 @@ dependencies = [
[[package]]
name = "arrow-buffer"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d18b89b4c4f4811d0858175e79541fe98e33e18db3b011708bc287b1240593f"
-dependencies = [
- "bytes",
- "half",
- "num-bigint",
- "num-traits",
-]
-
-[[package]]
-name = "arrow-buffer"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c6cd424c2693bcdbc150d843dc9d4d137dd2de4782ce6df491ad11a3a0416c0"
+checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5"
dependencies = [
"bytes",
"half",
@@ -648,39 +609,18 @@ dependencies = [
[[package]]
name = "arrow-cast"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "722b5c41dd1d14d0a879a1bce92c6fe33f546101bb2acce57a209825edd075b3"
-dependencies = [
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-ord 57.3.1",
- "arrow-schema 57.3.1",
- "arrow-select 57.3.1",
- "atoi",
- "base64",
- "chrono",
- "half",
- "lexical-core",
- "num-traits",
- "ryu",
-]
-
-[[package]]
-name = "arrow-cast"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c5aefb56a2c02e9e2b30746241058b85f8983f0fcff2ba0c6d09006e1cded7f"
+checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34"
dependencies = [
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-ord 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-ord",
+ "arrow-schema",
+ "arrow-select",
"atoi",
- "base64",
+ "base64 0.22.1",
"chrono",
"half",
"lexical-core",
@@ -690,28 +630,13 @@ dependencies = [
[[package]]
name = "arrow-csv"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "27ddb80a4848e03b1655af496d5ac2563a779e5742fcb48f2ca2e089c9cd2197"
-dependencies = [
- "arrow-array 57.3.1",
- "arrow-cast 57.3.1",
- "arrow-schema 57.3.1",
- "chrono",
- "csv",
- "csv-core",
- "regex",
-]
-
-[[package]]
-name = "arrow-csv"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e94e8cf7e517657a52b91ea1263acf38c4ca62a84655d72458a3359b12ab97de"
+checksum = "af0dd6d90d1955e9f9a014c1e563ee8aeffc21909085d25623e1da44d96eca26"
dependencies = [
- "arrow-array 58.3.0",
- "arrow-cast 58.3.0",
- "arrow-schema 58.3.0",
+ "arrow-array",
+ "arrow-cast",
+ "arrow-schema",
"chrono",
"csv",
"csv-core",
@@ -720,69 +645,43 @@ dependencies = [
[[package]]
name = "arrow-data"
-version = "57.3.1"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c1683705c63dcf0d18972759eda48489028cbbff67af7d6bef2c6b7b74ab778a"
+checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b"
dependencies = [
- "arrow-buffer 57.3.1",
- "arrow-schema 57.3.1",
+ "arrow-buffer",
+ "arrow-schema",
"half",
"num-integer",
"num-traits",
]
-[[package]]
-name = "arrow-data"
-version = "58.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c88210023a2bfee1896af366309a3028fc3bcbd6515fa29a7990ee1baa08ee0"
-dependencies = [
- "arrow-buffer 58.3.0",
- "arrow-schema 58.3.0",
- "half",
- "num-integer",
- "num-traits",
-]
-
-[[package]]
-name = "arrow-ipc"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8cf72d04c07229fbf4dbebe7145cac37d7cf7ec582fe705c6b92cb314af096ab"
-dependencies = [
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
- "arrow-select 57.3.1",
- "flatbuffers",
-]
-
[[package]]
name = "arrow-ipc"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "238438f0834483703d88896db6fe5a7138b2230debc31b34c0336c2996e3c64f"
+checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b"
dependencies = [
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
+ "arrow-select",
"flatbuffers",
]
[[package]]
name = "arrow-json"
-version = "57.3.1"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a84a905f41fedfcd7679813c89a61dc369c0f932b27aa8dcc6aa051cc781a97d"
+checksum = "b8a96aed3931c076adee39ec2a40d8219fc7f09e79bcdaca1df16272993e1e14"
dependencies = [
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-cast 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-cast",
+ "arrow-ord",
+ "arrow-schema",
+ "arrow-select",
"chrono",
"half",
"indexmap 2.14.0",
@@ -796,94 +695,37 @@ dependencies = [
"simdutf8",
]
-[[package]]
-name = "arrow-json"
-version = "58.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "205ca2119e6d679d5c133c6f30e68f027738d95ed948cf77677ea69c7800036b"
-dependencies = [
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-cast 58.3.0",
- "arrow-ord 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
- "chrono",
- "half",
- "indexmap 2.14.0",
- "itoa",
- "lexical-core",
- "memchr",
- "num-traits",
- "ryu",
- "serde_core",
- "serde_json",
- "simdutf8",
-]
-
-[[package]]
-name = "arrow-ord"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "082342947d4e5a2bcccf029a0a0397e21cb3bb8421edd9571d34fb5dd2670256"
-dependencies = [
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
- "arrow-select 57.3.1",
-]
-
[[package]]
name = "arrow-ord"
-version = "58.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1bffd8fd2579286a5d63bac898159873e5094a79009940bcb42bbfce4f19f1d0"
-dependencies = [
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
-]
-
-[[package]]
-name = "arrow-row"
-version = "57.3.1"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e3a931b520a2a5e22033e01a6f2486b4cdc26f9106b759abeebc320f125e94d7"
+checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0"
dependencies = [
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
- "half",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
+ "arrow-select",
]
[[package]]
name = "arrow-row"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bab5994731204603c73ba69267616c50f80780774c6bb0476f1f830625115e0c"
+checksum = "514ba0ef0d4c5896202dae736251ce415abb43a950bed570fb7981b8716c0e4c"
dependencies = [
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-schema 58.3.0",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
"half",
]
[[package]]
name = "arrow-schema"
-version = "57.3.1"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e4cf0d4a6609679e03002167a61074a21d7b1ad9ea65e462b2c0a97f8a3b2bc6"
-
-[[package]]
-name = "arrow-schema"
-version = "58.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f633dbfdf39c039ada1bf9e34c694816eb71fbb7dc78f613993b7245e078a1ed"
+checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e"
dependencies = [
"bitflags 2.13.1",
"serde",
@@ -892,60 +734,29 @@ dependencies = [
[[package]]
name = "arrow-select"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b320d86a9806923663bb0fd9baa65ecaba81cb0cd77ff8c1768b9716b4ef891"
-dependencies = [
- "ahash 0.8.12",
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
- "num-traits",
-]
-
-[[package]]
-name = "arrow-select"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8cd065c54172ac787cf3f2f8d4107e0d3fdc26edba76fdf4f4cc170258942222"
+checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad"
dependencies = [
"ahash 0.8.12",
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-schema 58.3.0",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
"num-traits",
]
[[package]]
name = "arrow-string"
-version = "57.3.1"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b493e99162e5764077e7823e50ba284858d365922631c7aaefe9487b1abd02c2"
+checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c"
dependencies = [
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-data 57.3.1",
- "arrow-schema 57.3.1",
- "arrow-select 57.3.1",
- "memchr",
- "num-traits",
- "regex",
- "regex-syntax",
-]
-
-[[package]]
-name = "arrow-string"
-version = "58.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "29dd7cda3ab9692f43a2e4acc444d760cc17b12bb6d8232ddf64e9bab7c06b42"
-dependencies = [
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
+ "arrow-select",
"memchr",
"num-traits",
"regex",
@@ -1333,7 +1144,7 @@ version = "0.30.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "16e2cdb6d5ed835199484bb92bb8b3edd526effe995c61732580439c1a67e2e9"
dependencies = [
- "base64",
+ "base64 0.22.1",
"http 1.4.2",
"log",
"rustls",
@@ -1966,6 +1777,12 @@ version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+[[package]]
+name = "base64"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
+
[[package]]
name = "base64-simd"
version = "0.8.0"
@@ -2294,7 +2111,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee04c4c84f1f811b017f2fbb7dd8815c976e7ca98593de9c1e2afad0f636bff4"
dependencies = [
"async-stream",
- "base64",
+ "base64 0.22.1",
"bitflags 2.13.1",
"bollard-buildkit-proto",
"bollard-stubs",
@@ -2351,7 +2168,7 @@ version = "1.52.1-rc.29.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f0a8ca8799131c1837d1282c3f81f31e76ceb0ce426e04a7fe1ccee3287c066"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bollard-buildkit-proto",
"bytes",
"prost",
@@ -2447,7 +2264,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7969a9ba84b0ff843813e7249eed1678d9b6607ce5a3b8f0a47af3fcf7978e6e"
dependencies = [
"ahash 0.8.12",
- "base64",
+ "base64 0.22.1",
"bitvec",
"getrandom 0.2.17",
"getrandom 0.3.4",
@@ -2492,7 +2309,7 @@ version = "0.22.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2235eb320cd7178862a32dd111bd0c0f71a368e393add4914c50129add478eab"
dependencies = [
- "arrow 58.3.0",
+ "arrow",
"buoyant_kernel_derive",
"bytes",
"chrono",
@@ -2501,7 +2318,7 @@ dependencies = [
"indexmap 2.14.0",
"itertools 0.14.0",
"object_store",
- "parquet 58.3.0",
+ "parquet",
"percent-encoding",
"rand 0.9.5",
"reqwest 0.13.4",
@@ -3605,6 +3422,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
dependencies = [
"generic-array",
+ "rand_core 0.6.4",
"typenum",
]
@@ -3667,6 +3485,15 @@ version = "0.0.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a949c44fcacbbbb7ada007dc7acb34603dd97cd47de5d054f2b6493ecebb483"
+[[package]]
+name = "ctr"
+version = "0.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835"
+dependencies = [
+ "cipher 0.4.4",
+]
+
[[package]]
name = "ctr"
version = "0.10.1"
@@ -3787,7 +3614,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5f8c8abaf78cfe3cb7838d7cfcf75e4e3bf02fe3e54571e0a1f10bb485cf8fbf"
dependencies = [
"async-stream",
- "base64",
+ "base64 0.22.1",
"cfg-if",
"cfg_aliases",
"compio",
@@ -4091,17 +3918,17 @@ version = "0.32.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4588e95ff3b2ccdba56d9ec262bd3467c0593000f729402528706f62be8be1ca"
dependencies = [
- "arrow 58.3.0",
- "arrow-arith 58.3.0",
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-cast 58.3.0",
- "arrow-ipc 58.3.0",
- "arrow-json 58.3.0",
- "arrow-ord 58.3.0",
- "arrow-row 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
+ "arrow",
+ "arrow-arith",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-cast",
+ "arrow-ipc",
+ "arrow-json",
+ "arrow-ord",
+ "arrow-row",
+ "arrow-schema",
+ "arrow-select",
"async-trait",
"buoyant_kernel",
"bytes",
@@ -4118,7 +3945,7 @@ dependencies = [
"num_cpus",
"object_store",
"parking_lot",
- "parquet 58.3.0",
+ "parquet",
"percent-encoding",
"percent-encoding-rfc3986",
"pin-project-lite",
@@ -4526,7 +4353,7 @@ version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29547a1dc60885a552306986316bc9701ba120c1a8db6769fa68691529ad373d"
dependencies = [
- "base64",
+ "base64 0.22.1",
"serde",
"serde_json",
]
@@ -4638,7 +4465,7 @@ version = "9.1.0-alpha.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "12bb303aa6e1d28c0c86b6fbfe484fd0fd3f512629aeed1ac4f6b85f81d9834a"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"dyn-clone",
"flate2",
@@ -5384,13 +5211,23 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "ghash"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1"
+dependencies = [
+ "opaque-debug",
+ "polyval 0.6.2",
+]
+
[[package]]
name = "ghash"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2eecf2d5dc9b66b732b97707a0210906b1d30523eb773193ab777c0c84b3e8d5"
dependencies = [
- "polyval",
+ "polyval 0.7.3",
]
[[package]]
@@ -6390,7 +6227,7 @@ version = "0.1.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"futures-channel",
"futures-util",
@@ -6450,25 +6287,26 @@ dependencies = [
[[package]]
name = "iceberg"
-version = "0.9.1"
+version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d9c3fc1f55c84ff64645c0d2ee35159f5574d33f729fc0860783bc247c8b8c5"
+checksum = "6e50fb53c7480f414911ab76b6e14c0042d54ad5c30f6ac2bf2f52a487a25716"
dependencies = [
+ "aes-gcm 0.10.3",
"anyhow",
"apache-avro",
"array-init",
- "arrow-arith 57.3.1",
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-cast 57.3.1",
- "arrow-ord 57.3.1",
- "arrow-schema 57.3.1",
- "arrow-select 57.3.1",
- "arrow-string 57.3.1",
+ "arrow-arith",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-cast",
+ "arrow-ord",
+ "arrow-schema",
+ "arrow-select",
+ "arrow-string",
"as-any",
"async-trait",
"backon",
- "base64",
+ "base64 0.22.1",
"bimap",
"bytes",
"chrono",
@@ -6483,7 +6321,7 @@ dependencies = [
"murmur3",
"once_cell",
"ordered-float 4.6.0",
- "parquet 57.3.1",
+ "parquet",
"rand 0.9.5",
"reqwest 0.12.28",
"roaring",
@@ -6495,18 +6333,20 @@ dependencies = [
"serde_with",
"strum 0.27.2",
"tokio",
+ "tracing",
"typed-builder 0.20.1",
"typetag",
"url",
"uuid",
+ "zeroize",
"zstd",
]
[[package]]
name = "iceberg-catalog-rest"
-version = "0.9.1"
+version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a49dfef578060c3a2a3f619522dcb25382a7ce85336dcb500495d4b8d72ae714"
+checksum = "d1500a26a9b18f286a319e914ce7290dddb7e2eef810edf535afc4ce0c8a6f45"
dependencies = [
"async-trait",
"chrono",
@@ -6518,25 +6358,25 @@ dependencies = [
"serde_derive",
"serde_json",
"tokio",
- "tracing",
"typed-builder 0.20.1",
"uuid",
]
[[package]]
name = "iceberg-storage-opendal"
-version = "0.9.1"
+version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "30acae4698949eea6cfbd3e64ddf1c3ec39e6b7191acb0b0da2dcd1a25dfa842"
+checksum = "983efbae430ed40c074af69a8b0aff99bee73e54686cf26ecde97f84127812a2"
dependencies = [
"anyhow",
"async-trait",
"bytes",
"cfg-if",
+ "futures",
"iceberg",
"opendal",
- "reqsign",
- "reqwest 0.12.28",
+ "reqsign-aws-v4",
+ "reqsign-core",
"serde",
"typetag",
"url",
@@ -6905,10 +6745,10 @@ dependencies = [
name = "iggy_common"
version = "0.11.0-edge.6"
dependencies = [
- "aes-gcm",
+ "aes-gcm 0.11.0",
"async-broadcast",
"async-trait",
- "base64",
+ "base64 0.22.1",
"blake3",
"bon",
"byte-unit",
@@ -6976,7 +6816,7 @@ name = "iggy_connector_doris_sink"
version = "0.2.0-edge.4"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"blake3",
"bytes",
"humantime",
@@ -6996,7 +6836,7 @@ name = "iggy_connector_elasticsearch_sink"
version = "0.5.0-edge.4"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"dashmap",
"elasticsearch",
"iggy_common",
@@ -7032,7 +6872,7 @@ name = "iggy_connector_http_sink"
version = "0.5.0-edge.4"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"bytes",
"humantime",
"iggy_connector_sdk",
@@ -7053,15 +6893,15 @@ dependencies = [
name = "iggy_connector_iceberg_sink"
version = "0.5.0-edge.4"
dependencies = [
- "arrow-array 57.3.1",
- "arrow-json 57.3.1",
+ "arrow-array",
+ "arrow-json",
"async-trait",
"dashmap",
"iceberg",
"iceberg-catalog-rest",
"iceberg-storage-opendal",
"iggy_connector_sdk",
- "parquet 57.3.1",
+ "parquet",
"serde",
"simd-json",
"strum 0.28.0",
@@ -7076,7 +6916,7 @@ version = "0.5.0-edge.4"
dependencies = [
"async-trait",
"axum",
- "base64",
+ "base64 0.22.1",
"bytes",
"iggy_common",
"iggy_connector_sdk",
@@ -7098,7 +6938,7 @@ dependencies = [
"ahash 0.8.12",
"async-trait",
"axum",
- "base64",
+ "base64 0.22.1",
"chrono",
"csv",
"dashmap",
@@ -7122,7 +6962,7 @@ name = "iggy_connector_meilisearch_sink"
version = "0.5.0-edge.4"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"iggy_common",
"iggy_connector_sdk",
"meilisearch-sdk",
@@ -7175,7 +7015,7 @@ name = "iggy_connector_postgres_source"
version = "0.5.0-edge.4"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"chrono",
"dashmap",
"futures",
@@ -7227,13 +7067,13 @@ dependencies = [
name = "iggy_connector_redshift_sink"
version = "0.4.1-edge.1"
dependencies = [
- "arrow 57.3.1",
+ "arrow",
"async-trait",
"chrono",
"humantime",
"iggy_common",
"iggy_connector_sdk",
- "parquet 57.3.1",
+ "parquet",
"rust-s3",
"secrecy",
"serde",
@@ -7250,7 +7090,7 @@ name = "iggy_connector_s3_sink"
version = "0.5.0-edge.4"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"byte-unit",
"chrono",
"dashmap",
@@ -7273,7 +7113,7 @@ dependencies = [
"anyhow",
"apache-avro",
"async-trait",
- "base64",
+ "base64 0.22.1",
"dashmap",
"flatbuffers",
"http 1.4.2",
@@ -7319,7 +7159,7 @@ name = "iggy_connector_surrealdb_sink"
version = "0.5.0-edge.4"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"bytes",
"iggy_common",
"iggy_connector_sdk",
@@ -7525,10 +7365,10 @@ checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02"
name = "integration"
version = "0.0.1"
dependencies = [
- "arrow 57.3.1",
+ "arrow",
"assert_cmd",
"async-trait",
- "base64",
+ "base64 0.22.1",
"bon",
"bytemuck",
"bytes",
@@ -7557,7 +7397,7 @@ dependencies = [
"lazy_static",
"libc",
"mongodb",
- "parquet 57.3.1",
+ "parquet",
"pgwire",
"predicates",
"rand 0.10.2",
@@ -7693,10 +7533,12 @@ dependencies = [
"jiff-core",
"jiff-static",
"jiff-tzdb-platform",
+ "js-sys",
"log",
"portable-atomic",
"portable-atomic-util",
"serde_core",
+ "wasm-bindgen",
"windows-link 0.2.1",
]
@@ -7826,7 +7668,7 @@ version = "10.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc"
dependencies = [
- "base64",
+ "base64 0.22.1",
"ed25519-dalek",
"getrandom 0.2.17",
"hmac 0.12.1",
@@ -8357,15 +8199,6 @@ dependencies = [
"libc",
]
-[[package]]
-name = "lz4_flex"
-version = "0.12.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "90071f8077f8e40adfc4b7fe9cd495ce316263f19e75c2211eeff3fdf475a3d9"
-dependencies = [
- "twox-hash",
-]
-
[[package]]
name = "lz4_flex"
version = "0.13.1"
@@ -8494,6 +8327,16 @@ version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ebb8d8732c6a6df3d8f032a82911cfc747e00efb95cc46e8d0acd5b5b88570c"
+[[package]]
+name = "mea"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c709842c4ce65cb91e2666ad5319dfc1efc3af0d34f02075eddca9000d9f8afb"
+dependencies = [
+ "hashbrown 0.17.1",
+ "slab",
+]
+
[[package]]
name = "meilisearch-index-setting-macro"
version = "0.33.0"
@@ -8777,7 +8620,7 @@ version = "3.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b814038f367d212f55de0a630cb35102a9b8ca23785a86955d62c0087c93846d"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bitflags 2.13.1",
"bson",
"derive-where",
@@ -9202,7 +9045,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "622acbc9100d3c10e2ee15804b0caa40e55c933d5aa53814cd520805b7958a49"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"bytes",
"chrono",
"form_urlencoded",
@@ -9243,7 +9086,7 @@ checksum = "27be39870c558e1fbc5a5f8d4a29aa916268c1dbcb9d467f2a9bbab35598060e"
dependencies = [
"arc-swap",
"async-trait",
- "base64",
+ "base64 0.22.1",
"bytes",
"cargo_metadata",
"cfg-if",
@@ -9301,33 +9144,132 @@ version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
+[[package]]
+name = "opaque-debug"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
+
[[package]]
name = "opendal"
-version = "0.55.0"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "96c9c85ce253ff87225e7669979d877a20c98a06604ec9d6dd5f4473e08f1ae1"
+dependencies = [
+ "ctor 1.0.9",
+ "opendal-core",
+ "opendal-layer-concurrent-limit",
+ "opendal-layer-logging",
+ "opendal-layer-retry",
+ "opendal-layer-timeout",
+ "opendal-service-fs",
+ "opendal-service-s3",
+]
+
+[[package]]
+name = "opendal-core"
+version = "0.57.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d075ab8a203a6ab4bc1bce0a4b9fe486a72bf8b939037f4b78d95386384bc80a"
+checksum = "c4f8607c90e2c963a91467f50fb49fbc7fb3d573f88cea219ca59ccd3740b309"
dependencies = [
"anyhow",
- "backon",
- "base64",
+ "base64 0.22.1",
"bytes",
- "crc32c",
"futures",
- "getrandom 0.2.17",
"http 1.4.2",
"http-body 1.1.0",
"jiff",
"log",
- "md-5 0.10.6",
+ "md-5 0.11.0",
+ "mea",
"percent-encoding",
- "quick-xml 0.38.4",
- "reqsign",
- "reqwest 0.12.28",
+ "quick-xml 0.39.4",
+ "reqsign-core",
+ "reqwest 0.13.4",
"serde",
"serde_json",
"tokio",
"url",
"uuid",
+ "web-time",
+]
+
+[[package]]
+name = "opendal-layer-concurrent-limit"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d6f81ba6960e3fae1882f253b114b21d7e444e1534f209c7737a79f6243eb6f"
+dependencies = [
+ "futures",
+ "http 1.4.2",
+ "mea",
+ "opendal-core",
+]
+
+[[package]]
+name = "opendal-layer-logging"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "58ada45c6d81d1aa4c9305d0c7d4bc317c59c85866a0908a2d75a7a978aa5ee2"
+dependencies = [
+ "log",
+ "opendal-core",
+]
+
+[[package]]
+name = "opendal-layer-retry"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b2a25a718afb81fad81cb9a0580a1cb989221fa2317f888c6a37f8dad408eb7"
+dependencies = [
+ "backon",
+ "log",
+ "opendal-core",
+]
+
+[[package]]
+name = "opendal-layer-timeout"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e91f731724c213af81e9d03517859c8fc47b4578e64ad61ae4f099f10fe36e3"
+dependencies = [
+ "opendal-core",
+ "tokio",
+]
+
+[[package]]
+name = "opendal-service-fs"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22e89a665fef0e6bd249cf5ea47fc174b7ba892159bee4b9382528b1ca873a2c"
+dependencies = [
+ "bytes",
+ "log",
+ "opendal-core",
+ "serde",
+ "tokio",
+ "xattr",
+]
+
+[[package]]
+name = "opendal-service-s3"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "313d46c9f5ae70bca26b7c3e3fbb9b639292625f28af73aa016f47e788af9deb"
+dependencies = [
+ "base64 0.22.1",
+ "bytes",
+ "crc32c",
+ "http 1.4.2",
+ "log",
+ "md-5 0.11.0",
+ "opendal-core",
+ "quick-xml 0.39.4",
+ "reqsign-aws-v4",
+ "reqsign-core",
+ "reqsign-file-read-tokio",
+ "serde",
+ "url",
]
[[package]]
@@ -9547,54 +9489,18 @@ dependencies = [
[[package]]
name = "parquet"
-version = "57.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e832c6aa20310fc6de7ea5a3f4e20d34fd83e3b43229d32b81ffe5c14d74692"
-dependencies = [
- "ahash 0.8.12",
- "arrow-array 57.3.1",
- "arrow-buffer 57.3.1",
- "arrow-cast 57.3.1",
- "arrow-data 57.3.1",
- "arrow-ipc 57.3.1",
- "arrow-schema 57.3.1",
- "arrow-select 57.3.1",
- "base64",
- "brotli",
- "bytes",
- "chrono",
- "flate2",
- "futures",
- "half",
- "hashbrown 0.16.1",
- "lz4_flex 0.12.2",
- "num-bigint",
- "num-integer",
- "num-traits",
- "paste",
- "seq-macro",
- "simdutf8",
- "snap",
- "thrift",
- "tokio",
- "twox-hash",
- "zstd",
-]
-
-[[package]]
-name = "parquet"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5dafa7d01085b62a47dd0c1829550a0a36710ea9c4fe358a05a85477cec8a908"
+checksum = "d298093b2dec60289dce0684c986d0f7679e9dd15771c2c65406e1aaf604a704"
dependencies = [
"ahash 0.8.12",
- "arrow-array 58.3.0",
- "arrow-buffer 58.3.0",
- "arrow-data 58.3.0",
- "arrow-ipc 58.3.0",
- "arrow-schema 58.3.0",
- "arrow-select 58.3.0",
- "base64",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-ipc",
+ "arrow-schema",
+ "arrow-select",
+ "base64 0.22.1",
"brotli",
"bytes",
"chrono",
@@ -9608,6 +9514,7 @@ dependencies = [
"num-traits",
"object_store",
"paste",
+ "ring",
"seq-macro",
"simdutf8",
"snap",
@@ -9770,7 +9677,7 @@ version = "3.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be"
dependencies = [
- "base64",
+ "base64 0.22.1",
"serde_core",
]
@@ -9856,7 +9763,7 @@ checksum = "7981cfde34009be689a05a30c497ad5fbb552531d3d54230b3627264ff1bc384"
dependencies = [
"async-trait",
"aws-lc-rs",
- "base64",
+ "base64 0.22.1",
"bytes",
"chrono",
"derive-new",
@@ -10051,6 +9958,18 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "polyval"
+version = "0.6.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.2.17",
+ "opaque-debug",
+ "universal-hash 0.5.1",
+]
+
[[package]]
name = "polyval"
version = "0.7.3"
@@ -10059,7 +9978,7 @@ checksum = "f0fa31d631f2b2cb2a544d0aa321ce847a94764d701ca2becc411138b93d49cd"
dependencies = [
"cpubits",
"cpufeatures 0.3.0",
- "universal-hash",
+ "universal-hash 0.6.1",
]
[[package]]
@@ -10096,7 +10015,7 @@ version = "0.6.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514"
dependencies = [
- "base64",
+ "base64 0.22.1",
"byteorder",
"bytes",
"fallible-iterator",
@@ -10527,9 +10446,9 @@ checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
[[package]]
name = "quick-xml"
-version = "0.37.5"
+version = "0.38.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "331e97a1af0bf59823e6eadffe373d7b27f485be8748f71471c662c1f269b7fb"
+checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c"
dependencies = [
"memchr",
"serde",
@@ -10537,9 +10456,9 @@ dependencies = [
[[package]]
name = "quick-xml"
-version = "0.38.4"
+version = "0.39.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c"
+checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e"
dependencies = [
"memchr",
"serde",
@@ -10547,9 +10466,9 @@ dependencies = [
[[package]]
name = "quick-xml"
-version = "0.39.4"
+version = "0.41.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e"
+checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1"
dependencies = [
"memchr",
"serde",
@@ -10939,31 +10858,71 @@ dependencies = [
]
[[package]]
-name = "reqsign"
-version = "0.16.5"
+name = "reqsign-aws-core"
+version = "3.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43451dbf3590a7590684c25fb8d12ecdcc90ed3ac123433e500447c7d77ed701"
+checksum = "bac4749b7dfa7bfaccd01eb03e9dc795ed37e3f20d6f0f38e2c67ee85ad6bc86"
dependencies = [
- "anyhow",
- "async-trait",
- "base64",
- "chrono",
+ "bytes",
"form_urlencoded",
- "getrandom 0.2.17",
"hex",
- "hmac 0.12.1",
- "home",
"http 1.4.2",
"log",
"percent-encoding",
- "quick-xml 0.37.5",
- "rand 0.8.7",
- "reqwest 0.12.28",
+ "quick-xml 0.41.0",
+ "reqsign-core",
"rust-ini",
"serde",
"serde_json",
- "sha1 0.10.7",
- "sha2 0.10.9",
+ "serde_urlencoded",
+ "sha1 0.11.0",
+]
+
+[[package]]
+name = "reqsign-aws-v4"
+version = "3.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff250f0fd0b913fbd565e405acc553da0f13bde30bfb5403178c9d0313cdc15f"
+dependencies = [
+ "bytes",
+ "http 1.4.2",
+ "log",
+ "quick-xml 0.41.0",
+ "reqsign-aws-core",
+ "reqsign-core",
+ "serde",
+]
+
+[[package]]
+name = "reqsign-core"
+version = "3.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff052daffb0599681c50f85c59e7236438976efe991ab864edd9f3b235501a0f"
+dependencies = [
+ "anyhow",
+ "base64 0.23.1",
+ "bytes",
+ "futures",
+ "hex",
+ "hmac 0.13.0",
+ "http 1.4.2",
+ "jiff",
+ "log",
+ "mea",
+ "percent-encoding",
+ "sha1 0.11.0",
+ "sha2 0.11.0",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "reqsign-file-read-tokio"
+version = "3.0.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b3235df90a6bca681aa47dd86f2393d122a6d77042aa8a7c81e218cd45c5bfc0"
+dependencies = [
+ "anyhow",
+ "reqsign-core",
"tokio",
]
@@ -10973,7 +10932,7 @@ version = "0.12.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"futures-core",
"futures-util",
@@ -11016,7 +10975,7 @@ version = "0.13.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"encoding_rs",
"futures-channel",
@@ -11214,7 +11173,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "14db48ee17a9ba61810ab1a9c1beb7d06d8136ae39ac25a1137f10d357af01af"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"bytes",
"chrono",
"futures",
@@ -11376,7 +11335,7 @@ dependencies = [
"async-trait",
"aws-creds",
"aws-region",
- "base64",
+ "base64 0.22.1",
"bytes",
"cfg-if",
"futures-util",
@@ -11967,7 +11926,7 @@ version = "3.21.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bs58",
"chrono",
"hex",
@@ -12591,7 +12550,7 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"cfg-if",
"chrono",
@@ -12697,7 +12656,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e"
dependencies = [
"atoi",
- "base64",
+ "base64 0.22.1",
"bitflags 2.13.1",
"byteorder",
"chrono",
@@ -13641,7 +13600,7 @@ checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
dependencies = [
"async-trait",
"axum",
- "base64",
+ "base64 0.22.1",
"bytes",
"h2 0.4.15",
"http 1.4.2",
@@ -14175,6 +14134,16 @@ version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
+[[package]]
+name = "universal-hash"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea"
+dependencies = [
+ "crypto-common 0.1.7",
+ "subtle",
+]
+
[[package]]
name = "universal-hash"
version = "0.6.1"
@@ -14209,7 +14178,7 @@ version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0"
dependencies = [
- "base64",
+ "base64 0.22.1",
"flate2",
"log",
"percent-encoding",
@@ -14226,7 +14195,7 @@ version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c"
dependencies = [
- "base64",
+ "base64 0.22.1",
"http 1.4.2",
"httparse",
"log",
@@ -14257,7 +14226,7 @@ version = "0.45.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "80be9b06fbae3b8b303400ab20778c80bbaf338f563afe567cf3c9eea17b47ef"
dependencies = [
- "base64",
+ "base64 0.22.1",
"data-url",
"flate2",
"fontdb",
@@ -15173,7 +15142,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031"
dependencies = [
"assert-json-diff",
- "base64",
+ "base64 0.22.1",
"deadpool",
"futures",
"http 1.4.2",
diff --git a/Cargo.toml b/Cargo.toml
index 7872f4093..7293cdd79 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -96,9 +96,9 @@ apple-native-keyring-store = { version = "1.0.1", features = ["keychain"] }
# build graph happens to enable it via feature unification.
argon2 = { version = "0.5.3", features = ["std"] }
# Pinned to 57 because iceberg 0.9.1 still requires arrow/parquet 57.
-arrow = "57.3.1"
-arrow-array = "57.3.1"
-arrow-json = "57.3.1"
+arrow = "58.4.0"
+arrow-array = "58.4.0"
+arrow-json = "58.4.0"
assert_cmd = "2.2.2"
async-broadcast = "0.7.2"
async-channel = "2.5.0"
@@ -203,9 +203,9 @@ humantime = "2.4.0"
hwlocality = "1.0.0-alpha.12"
hyper = "1.11.0"
hyper-util = { version = "0.1.20", features = ["server-auto", "service"] }
-iceberg = "0.9.1"
-iceberg-catalog-rest = "0.9.1"
-iceberg-storage-opendal = "0.9.1"
+iceberg = "0.10.1"
+iceberg-catalog-rest = "0.10.1"
+iceberg-storage-opendal = "0.10.1"
iggy = { path = "core/sdk", version = "0.11.0-edge.6" }
iggy-cli = { path = "core/cli", version = "0.14.0-edge.6" }
iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.6" }
@@ -257,7 +257,7 @@ opentelemetry_sdk = { version = "0.32.1", features = [
"experimental_trace_batch_span_processor_with_async_runtime",
] }
papaya = "0.2.4"
-parquet = "57.3.1"
+parquet = "58.4.0"
partitions = { path = "core/partitions" }
passterm = "2.0.6"
paste = "1.0"
diff --git a/core/connectors/sinks/iceberg_sink/src/catalog.rs b/core/connectors/sinks/iceberg_sink/src/catalog.rs
index d6b696277..53d4043c5 100644
--- a/core/connectors/sinks/iceberg_sink/src/catalog.rs
+++ b/core/connectors/sinks/iceberg_sink/src/catalog.rs
@@ -48,7 +48,6 @@ async fn get_rest_catalog(
let storage_factory: Arc<dyn iceberg::io::StorageFactory> = match &config.store_class {
IcebergSinkStoreClass::S3 => Arc::new(OpenDalStorageFactory::S3 {
- configured_scheme: "s3".to_string(),
customized_credential_load: None,
}),
other => {
diff --git a/core/connectors/sinks/iceberg_sink/src/props.rs b/core/connectors/sinks/iceberg_sink/src/props.rs
index 4815c530c..9cd2e13b2 100644
--- a/core/connectors/sinks/iceberg_sink/src/props.rs
+++ b/core/connectors/sinks/iceberg_sink/src/props.rs
@@ -30,6 +30,9 @@ fn get_props_s3(config: &IcebergSinkConfig) -> Result<HashMap<String, String>, E
let mut props: HashMap<String, String> = HashMap::new();
props.insert("s3.region".to_string(), config.store_region.clone());
props.insert("s3.endpoint".to_string(), config.store_url.clone());
+ // iceberg-storage-opendal 0.10 defaults to virtual-host-style addressing,
+ // which breaks MinIO-style endpoints; keep the path-style behavior.
+ props.insert("s3.path-style-access".to_string(), "true".to_string());
match (&config.store_access_key_id, &config.store_secret_access_key) {
(Some(access_key_id), Some(secret_access_key)) => {
props.insert("s3.access-key-id".to_string(), access_key_id.clone());
@@ -73,6 +76,7 @@ mod tests {
let props = get_props_s3(&config).expect("Should succeed without credentials");
assert_eq!(props.get("s3.region").unwrap(), "us-east-1");
assert_eq!(props.get("s3.endpoint").unwrap(), "http://localhost:9000");
+ assert_eq!(props.get("s3.path-style-access").unwrap(), "true");
assert!(!props.contains_key("s3.access-key-id"));
assert!(!props.contains_key("s3.secret-access-key"));
}
diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
index 39fa7197e..da932b4eb 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
@@ -117,7 +117,7 @@ async fn write_data_files(
table: &Table,
messages_schema: Schema,
) -> Result<Vec<DataFile>, Error> {
- let location = DefaultLocationGenerator::new(table.metadata().clone()).map_err(|err| {
+ let location = DefaultLocationGenerator::new(table.metadata()).map_err(|err| {
error!(
"Failed to get location on table: {}. Error: {}",
table.metadata().uuid(),
@@ -340,15 +340,23 @@ pub trait Router: std::fmt::Debug + Sync + Send {
#[cfg(test)]
mod tests {
use super::*;
+ use iceberg::Runtime as IcebergRuntime;
use iceberg::spec::{
FormatVersion, Literal, NestedField, PrimitiveLiteral, PrimitiveType,
Schema as IcebergSchema, SortOrder, TableMetadataBuilder, Transform, Type,
UnboundPartitionSpec,
};
use std::collections::HashMap;
+ use std::sync::OnceLock;
const REGION_FIELD_ID: i32 = 2;
+ // Table keeps only tokio handles, so the runtime must outlive every table.
+ fn test_runtime() -> &'static tokio::runtime::Runtime {
+ static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
+ RUNTIME.get_or_init(|| tokio::runtime::Runtime::new().expect("Failed to create runtime"))
+ }
+
fn in_memory_table(partition_spec: UnboundPartitionSpec) -> Table {
let schema = IcebergSchema::builder()
.with_fields(vec![
@@ -379,6 +387,7 @@ mod tests {
.identifier(TableIdent::from_strs(["test", "events"]).expect("Failed to build ident"))
.metadata(metadata)
.file_io(FileIO::new_with_memory())
+ .runtime(IcebergRuntime::new(test_runtime()))
.build()
.expect("Failed to build table")
}
@@ -390,8 +399,7 @@ mod tests {
}
fn write(table: &Table, payloads: &[Payload]) -> Vec<DataFile> {
- tokio::runtime::Runtime::new()
- .expect("Failed to create runtime")
+ test_runtime()
.block_on(write_data_files(payloads, table, Schema::Json))
.expect("Failed to write data files")
}
@@ -448,13 +456,11 @@ mod tests {
let data_files = write(&table, &payloads);
let file_path = data_files[0].file_path().to_string();
- let exists_after_delete = tokio::runtime::Runtime::new()
- .expect("Failed to create runtime")
- .block_on(async {
- assert!(table.file_io().exists(&file_path).await.expect("exists"));
- delete_uncommitted_files(table.file_io(), &data_files).await;
- table.file_io().exists(&file_path).await.expect("exists")
- });
+ let exists_after_delete = test_runtime().block_on(async {
+ assert!(table.file_io().exists(&file_path).await.expect("exists"));
+ delete_uncommitted_files(table.file_io(), &data_files).await;
+ table.file_io().exists(&file_path).await.expect("exists")
+ });
assert!(!exists_after_delete);
}
(iggy) 01/03: fix(connectors): write iceberg sink data into real table partitions
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/iceberg-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 14649bae09c429ca6d4fca88251c72d209dd3ede
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Aug 26 10:28:13 2026 +0200
fix(connectors): write iceberg sink data into real table partitions
The Iceberg sink built its partition key from an empty
PartitionSpec::builder() with hardcoded zero literals, so every
record landed in one dummy partition (e.g. year=1970) whatever the
table's spec said, and a batch spanning several partitions was
never split.
Read the default partition spec from table metadata instead and
push every batch through one TableWriter: partitioned tables split
it with RecordBatchPartitionSplitter and fan out to one data file
writer per partition value, unpartitioned tables map it to the
single key of their spec (with one null per void field, which the
commit-time arity check requires). Files finalized by a failed
batch are deleted before the error is returned so nothing is left
orphaned on the object store; files are kept on commit failure
because the catalog may already have applied the commit.
Ports #3860 with its review feedback applied. Fixes #3853
---
Cargo.lock | 2 +
core/connectors/sinks/iceberg_sink/Cargo.toml | 4 +
core/connectors/sinks/iceberg_sink/README.md | 7 +
.../sinks/iceberg_sink/src/router/mod.rs | 413 +++++++++++++++------
.../tests/connectors/fixtures/iceberg/container.rs | 64 +++-
.../tests/connectors/fixtures/iceberg/mod.rs | 3 +-
core/integration/tests/connectors/fixtures/mod.rs | 3 +-
.../tests/connectors/iceberg/iceberg_sink.rs | 82 +++-
8 files changed, 465 insertions(+), 113 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 251f7f309..1eb6e89f3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7053,6 +7053,7 @@ dependencies = [
name = "iggy_connector_iceberg_sink"
version = "0.5.0-edge.4"
dependencies = [
+ "arrow-array 57.3.1",
"arrow-json 57.3.1",
"async-trait",
"dashmap",
@@ -7064,6 +7065,7 @@ dependencies = [
"serde",
"simd-json",
"strum 0.28.0",
+ "tokio",
"tracing",
"uuid",
]
diff --git a/core/connectors/sinks/iceberg_sink/Cargo.toml b/core/connectors/sinks/iceberg_sink/Cargo.toml
index ec6574214..9e5e21cd9 100644
--- a/core/connectors/sinks/iceberg_sink/Cargo.toml
+++ b/core/connectors/sinks/iceberg_sink/Cargo.toml
@@ -35,6 +35,7 @@ ignored = ["dashmap"]
crate-type = ["cdylib", "lib"]
[dependencies]
+arrow-array = { workspace = true }
arrow-json = { workspace = true }
async-trait = { workspace = true }
dashmap = { workspace = true }
@@ -49,5 +50,8 @@ strum = { workspace = true }
tracing = { workspace = true }
uuid = { workspace = true }
+[dev-dependencies]
+tokio = { workspace = true }
+
[lints]
workspace = true
diff --git a/core/connectors/sinks/iceberg_sink/README.md b/core/connectors/sinks/iceberg_sink/README.md
index c9079e713..2ebb4405f 100644
--- a/core/connectors/sinks/iceberg_sink/README.md
+++ b/core/connectors/sinks/iceberg_sink/README.md
@@ -81,6 +81,13 @@ Example:
- Namespace: `nyc`
- Table name: `users`
+## Partitioned Tables
+
+Data files follow the table's default partition spec. Each batch is split by the spec's transforms
+(`identity`, `bucket`, `truncate`, `year`, `month`, `day`, `hour`) and every partition value present
+in the batch gets at least one Parquet file under its partition path. Unpartitioned tables are
+written into a single file per batch.
+
## Source Compatibility
The Iceberg sink expects **flat JSON** where each top-level key maps directly to a column in the
diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
index 3f43b9401..39fa7197e 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
@@ -17,16 +17,16 @@
use crate::router::arrow_streamer::JsonArrowReader;
use crate::slice_user_table;
+use arrow_array::RecordBatch;
use arrow_json::ReaderBuilder;
use async_trait::async_trait;
use iceberg::TableIdent;
-use iceberg::arrow::schema_to_arrow_schema;
-use iceberg::spec::{
- Literal, PartitionKey, PartitionSpec, PrimitiveLiteral, PrimitiveType, Struct, StructType,
-};
+use iceberg::arrow::{RecordBatchPartitionSplitter, schema_to_arrow_schema};
+use iceberg::io::FileIO;
+use iceberg::spec::{DataFile, PartitionKey, Struct};
use iceberg::table::Table;
use iceberg::transaction::{ApplyTransactionAction, Transaction};
-use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
+use iceberg::writer::base_writer::data_file_writer::{DataFileWriter, DataFileWriterBuilder};
use iceberg::writer::file_writer::ParquetWriterBuilder;
use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder;
use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
@@ -36,10 +36,17 @@ use iceberg::{
};
use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
use parquet::file::properties::WriterProperties;
+use std::collections::HashMap;
+use std::collections::hash_map::Entry;
use std::sync::Arc;
use tracing::{error, warn};
use uuid::Uuid;
+type ParquetDataFileWriterBuilder =
+ DataFileWriterBuilder<ParquetWriterBuilder, DefaultLocationGenerator, DefaultFileNameGenerator>;
+type ParquetDataFileWriter =
+ DataFileWriter<ParquetWriterBuilder, DefaultLocationGenerator, DefaultFileNameGenerator>;
+
fn format_error_chain(err: &dyn std::error::Error) -> String {
let mut chain = err.to_string();
let mut source = err.source();
@@ -67,55 +74,49 @@ async fn table_exists(route_field_val: &str, catalog: &dyn Catalog) -> Option<Ta
catalog.load_table(&table_ident).await.ok()
}
-pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
- match pt {
- PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
- PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
- PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
- PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
- PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
- PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
- PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
- PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
- PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
- PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
- PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
- PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
- PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
- PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
- _ => {
- error!("Partition type not supported");
- Err(Error::InvalidConfig)
- }
- }
-}
-
-fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
- let mut fields: Vec<Option<Literal>> = Vec::new();
+async fn write_data(
+ messages: &[Payload],
+ table: &Table,
+ catalog: &dyn Catalog,
+ messages_schema: Schema,
+) -> Result<(), Error> {
+ let data_files = write_data_files(messages, table, messages_schema).await?;
- if default_partition_type.fields().is_empty() {
- return Ok(None);
- };
+ // Data files are kept on commit failure: the catalog may have applied the
+ // commit before the error surfaced, and retries must check that first.
+ let table_commit = Transaction::new(table);
- for field in default_partition_type.fields() {
- let field_type = field.field_type.as_primitive_type().ok_or_else(|| {
- error!("The partition type of the configured iceberg table is not a primitive type");
- Error::InvalidConfig
- })?;
+ let action = table_commit.fast_append().add_data_files(data_files);
- let value = Some(Literal::Primitive(primitive_type_to_literal(field_type)?));
+ let tx = action.apply(table_commit).map_err(|err| {
+ let chain = format_error_chain(&err);
+ error!(
+ "Failed to apply transaction on table with UUID: {}, Error: {}",
+ table.metadata().uuid(),
+ chain
+ );
+ Error::TransactionApplyError(chain)
+ })?;
- fields.push(value);
- }
- Ok(Some(Struct::from_iter(fields)))
+ let _table = tx.commit(catalog).await.map_err(|err| {
+ let chain = format_error_chain(&err);
+ error!(
+ "Failed to commit transaction on table with UUID: {}, Error: {}",
+ table.metadata().uuid(),
+ chain
+ );
+ Error::CatalogCommitError(chain)
+ })?;
+ Ok(())
}
-async fn write_data(
+/// Writes the JSON payloads as Parquet data files under the table location,
+/// laid out by the table's default partition spec. Nothing is committed here.
+async fn write_data_files(
messages: &[Payload],
table: &Table,
- catalog: &dyn Catalog,
messages_schema: Schema,
-) -> Result<(), Error> {
+) -> Result<Vec<DataFile>, Error> {
let location = DefaultLocationGenerator::new(table.metadata().clone()).map_err(|err| {
error!(
"Failed to get location on table: {}. Error: {}",
@@ -145,27 +146,6 @@ async fn write_data(
let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);
- let partition_spec = PartitionSpec::builder(table.current_schema_ref());
-
- let partition_type = get_partition_type_value(table.metadata().default_partition_type())?;
-
- let mut writer = data_file_writer_builder
- .build(match partition_type {
- None => None,
- Some(p_type) => Some(PartitionKey::new(
- partition_spec
- .build()
- .map_err(|err| Error::InitError(err.to_string()))?,
- table.current_schema_ref(),
- p_type,
- )),
- })
- .await
- .map_err(|err| {
- error!("Error while constructing data file writer: {}", err);
- Error::InitError(err.to_string())
- })?;
-
let msgs: Vec<&simd_json::OwnedValue> = messages
.iter()
.filter_map(|payload| match payload {
@@ -201,6 +181,12 @@ async fn write_data(
Error::InitError(err.to_string())
})?;
+ let mut writer = TableWriter::new(table, data_file_writer_builder).map_err(|err| {
+ let chain = format_error_chain(&err);
+ error!("Error while constructing data file writer: {}", chain);
+ Error::InitError(chain)
+ })?;
+
let write_result: Result<(), Error> = async {
for batch in reader {
let batch_data = batch.map_err(|err| {
@@ -208,60 +194,138 @@ async fn write_data(
error!("Error while getting record batch: {}", chain);
Error::InvalidRecordValue(chain)
})?;
- writer.write(batch_data).await.map_err(|err| {
- let chain = format_error_chain(&err);
- error!("Error while writing record batch: {}", chain);
- Error::WriteFailure(chain)
- })?;
+ writer.write(batch_data).await?;
}
Ok(())
}
.await;
- if let Err(e) = &write_result {
+ let (data_files, close_error) = writer.close().await;
+ if let Some(err) = write_result.err().or(close_error) {
error!(
- "Batch loop failed ({}), closing writer to release resources",
- e
+ "Writing data files failed ({}), deleting the uncommitted ones",
+ err
);
- if let Err(close_err) = writer.close().await {
- error!("Failed to close writer after batch error: {}", close_err);
+ delete_uncommitted_files(table.file_io(), &data_files).await;
+ return Err(err);
+ }
+ Ok(data_files)
+}
+
+/// Files finalized by a failed batch are never committed, so they are removed
+/// instead of being left as orphans on the object store.
+async fn delete_uncommitted_files(file_io: &FileIO, data_files: &[DataFile]) {
+ for data_file in data_files {
+ if let Err(err) = file_io.delete(data_file.file_path()).await {
+ warn!(
+ "Failed to delete uncommitted data file {}: {}",
+ data_file.file_path(),
+ err
+ );
}
- return Err(write_result.unwrap_err());
}
+}
- let data_files = writer.close().await.map_err(|err| {
- let chain = format_error_chain(&err);
- error!(
- "Error while writing data records to Parquet file: {}",
- chain
- );
- Error::WriteFailure(chain)
- })?;
+/// Data file writer bound to the table's default partition spec.
+///
+/// Every batch is mapped to partition keys and fanned out to one data file
+/// writer per key, so a mixed batch lands in the right partitions. An
+/// unpartitioned table maps every batch to the single key of its spec.
+///
+/// The per-partition writers are owned here rather than by iceberg's
+/// `FanoutWriter` because its `close` stops at the first failing writer and
+/// loses the files the other writers already finalized.
+struct TableWriter {
+ partitioner: Partitioner,
+ builder: ParquetDataFileWriterBuilder,
+ writers: HashMap<Struct, ParquetDataFileWriter>,
+}
- let table_commit = Transaction::new(table);
+enum Partitioner {
+ Unpartitioned(PartitionKey),
+ /// Boxed to keep the enum small (`clippy::large_enum_variant`).
+ Partitioned(Box<RecordBatchPartitionSplitter>),
+}
- let action = table_commit.fast_append().add_data_files(data_files);
+impl TableWriter {
+ fn new(table: &Table, builder: ParquetDataFileWriterBuilder) -> iceberg::Result<Self> {
+ let partition_spec = table.metadata().default_partition_spec();
+ let schema = table.metadata().current_schema().clone();
+ let partitioner = if partition_spec.is_unpartitioned() {
+ // A spec whose fields are all `Transform::Void` counts as unpartitioned,
+ // but its partition type still has one field per void field and the
+ // commit rejects data files whose partition value has a different arity.
+ let void_field_count = table.metadata().default_partition_type().fields().len();
+ Partitioner::Unpartitioned(PartitionKey::new(
+ partition_spec.as_ref().clone(),
+ schema,
+ Struct::from_iter(vec![None; void_field_count]),
+ ))
+ } else {
+ let splitter = RecordBatchPartitionSplitter::try_new_with_computed_values(
+ schema,
+ partition_spec.clone(),
+ )?;
+ Partitioner::Partitioned(Box::new(splitter))
+ };
+ Ok(Self {
+ partitioner,
+ builder,
+ writers: HashMap::new(),
+ })
+ }
- let tx = action.apply(table_commit).map_err(|err| {
- let chain = format_error_chain(&err);
- error!(
- "Failed to apply transaction on table with UUID: {}, Error: {}",
- table.metadata().uuid(),
- chain
- );
- Error::TransactionApplyError(chain)
- })?;
+ async fn write(&mut self, batch: RecordBatch) -> Result<(), Error> {
+ let partitioned_batches = match &self.partitioner {
+ Partitioner::Unpartitioned(partition_key) => vec![(partition_key.clone(), batch)],
+ Partitioner::Partitioned(splitter) => splitter.split(&batch).map_err(|err| {
+ let chain = format_error_chain(&err);
+ error!("Error while splitting record batch by partition: {}", chain);
+ Error::InvalidRecordValue(chain)
+ })?,
+ };
+ for (partition_key, partition_batch) in partitioned_batches {
+ let writer = match self.writers.entry(partition_key.data().clone()) {
+ Entry::Occupied(entry) => entry.into_mut(),
+ Entry::Vacant(entry) => {
+ let writer = self
+ .builder
+ .build(Some(partition_key))
+ .await
+ .map_err(write_failure)?;
+ entry.insert(writer)
+ }
+ };
+ writer.write(partition_batch).await.map_err(write_failure)?;
+ }
+ Ok(())
+ }
- let _table = tx.commit(catalog).await.map_err(|err| {
- let chain = format_error_chain(&err);
- error!(
- "Failed to commit transaction on table with UUID: {}, Error: {}",
- table.metadata().uuid(),
- chain
- );
- Error::CatalogCommitError(chain)
- })?;
- Ok(())
+ /// Closes every partition writer even after one of them fails, so the
+ /// caller learns about all finalized files and can commit or delete them.
+ /// Returns the first close failure alongside the files.
+ async fn close(self) -> (Vec<DataFile>, Option<Error>) {
+ let mut data_files = Vec::new();
+ let mut first_error = None;
+ for mut writer in self.writers.into_values() {
+ match writer.close().await {
+ Ok(files) => data_files.extend(files),
+ Err(err) => {
+ let close_error = write_failure(err);
+ if first_error.is_none() {
+ first_error = Some(close_error);
+ }
+ }
+ }
+ }
+ (data_files, first_error)
+ }
+}
+
+fn write_failure(err: iceberg::Error) -> Error {
+ let chain = format_error_chain(&err);
+ error!("Error while writing data file: {}", chain);
+ Error::WriteFailure(chain)
}
#[async_trait]
@@ -272,3 +336,140 @@ pub trait Router: std::fmt::Debug + Sync + Send {
messages: Vec<ConsumedMessage>,
) -> Result<(), crate::Error>;
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iceberg::spec::{
+ FormatVersion, Literal, NestedField, PrimitiveLiteral, PrimitiveType,
+ Schema as IcebergSchema, SortOrder, TableMetadataBuilder, Transform, Type,
+ UnboundPartitionSpec,
+ };
+ use std::collections::HashMap;
+
+ const REGION_FIELD_ID: i32 = 2;
+
+ fn in_memory_table(partition_spec: UnboundPartitionSpec) -> Table {
+ let schema = IcebergSchema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
+ NestedField::required(
+ REGION_FIELD_ID,
+ "region",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
+ ])
+ .build()
+ .expect("Failed to build schema");
+ let metadata = TableMetadataBuilder::new(
+ schema,
+ partition_spec,
+ SortOrder::unsorted_order(),
+ "memory://warehouse/test/events".to_string(),
+ FormatVersion::V2,
+ HashMap::new(),
+ )
+ .expect("Failed to create table metadata")
+ .build()
+ .expect("Failed to build table metadata")
+ .metadata;
+
+ Table::builder()
+ .identifier(TableIdent::from_strs(["test", "events"]).expect("Failed to build ident"))
+ .metadata(metadata)
+ .file_io(FileIO::new_with_memory())
+ .build()
+ .expect("Failed to build table")
+ }
+
+ fn json_payloads(rows: &[(i64, &str)]) -> Vec<Payload> {
+ rows.iter()
+ .map(|(id, region)| Payload::Json(simd_json::json!({ "id": *id, "region": *region })))
+ .collect()
+ }
+
+ fn write(table: &Table, payloads: &[Payload]) -> Vec<DataFile> {
+ tokio::runtime::Runtime::new()
+ .expect("Failed to create runtime")
+ .block_on(write_data_files(payloads, table, Schema::Json))
+ .expect("Failed to write data files")
+ }
+
+ fn partition_region(data_file: &DataFile) -> String {
+ match &data_file.partition()[0] {
+ Some(Literal::Primitive(PrimitiveLiteral::String(region))) => region.clone(),
+ other => panic!("Expected string partition value, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn given_partitioned_table_should_write_one_data_file_per_partition_value() {
+ let partition_spec = UnboundPartitionSpec::builder()
+ .add_partition_field(REGION_FIELD_ID, "region", Transform::Identity)
+ .expect("Failed to add partition field")
+ .build();
+ let table = in_memory_table(partition_spec);
+ let payloads = json_payloads(&[(1, "eu"), (2, "us"), (3, "eu")]);
+
+ let data_files = write(&table, &payloads);
+
+ assert_eq!(data_files.len(), 2);
+ let by_region: HashMap<String, &DataFile> = data_files
+ .iter()
+ .map(|data_file| (partition_region(data_file), data_file))
+ .collect();
+ assert_eq!(by_region["eu"].record_count(), 2);
+ assert!(by_region["eu"].file_path().contains("/data/region=eu/"));
+ assert_eq!(by_region["us"].record_count(), 1);
+ assert!(by_region["us"].file_path().contains("/data/region=us/"));
+ }
+
+ #[test]
+ fn given_void_only_partition_spec_should_write_null_partition_values() {
+ let partition_spec = UnboundPartitionSpec::builder()
+ .add_partition_field(REGION_FIELD_ID, "region_void", Transform::Void)
+ .expect("Failed to add partition field")
+ .build();
+ let table = in_memory_table(partition_spec);
+ let payloads = json_payloads(&[(1, "eu"), (2, "us")]);
+
+ let data_files = write(&table, &payloads);
+
+ assert_eq!(data_files.len(), 1);
+ assert_eq!(data_files[0].partition().fields(), &[None]);
+ assert!(!data_files[0].file_path().contains("region_void="));
+ }
+
+ #[test]
+ fn given_uncommitted_files_should_be_deleted_from_store() {
+ let table = in_memory_table(UnboundPartitionSpec::builder().build());
+ let payloads = json_payloads(&[(1, "eu")]);
+ let data_files = write(&table, &payloads);
+ let file_path = data_files[0].file_path().to_string();
+
+ let exists_after_delete = tokio::runtime::Runtime::new()
+ .expect("Failed to create runtime")
+ .block_on(async {
+ assert!(table.file_io().exists(&file_path).await.expect("exists"));
+ delete_uncommitted_files(table.file_io(), &data_files).await;
+ table.file_io().exists(&file_path).await.expect("exists")
+ });
+
+ assert!(!exists_after_delete);
+ }
+
+ #[test]
+ fn given_unpartitioned_table_should_write_single_data_file_without_partition() {
+ let table = in_memory_table(UnboundPartitionSpec::builder().build());
+ let payloads = json_payloads(&[(1, "eu"), (2, "us")]);
+
+ let data_files = write(&table, &payloads);
+
+ assert_eq!(data_files.len(), 1);
+ assert_eq!(data_files[0].record_count(), 2);
+ assert!(data_files[0].partition().fields().is_empty());
+ assert!(data_files[0].file_path().contains("/data/"));
+ assert!(!data_files[0].file_path().contains("region="));
+ }
+}
diff --git a/core/integration/tests/connectors/fixtures/iceberg/container.rs b/core/integration/tests/connectors/fixtures/iceberg/container.rs
index cf5386c8b..71332f0ab 100644
--- a/core/integration/tests/connectors/fixtures/iceberg/container.rs
+++ b/core/integration/tests/connectors/fixtures/iceberg/container.rs
@@ -275,10 +275,11 @@ pub trait IcebergOps: Sync {
&self,
namespace: &str,
table: &str,
+ partition_spec: Option<serde_json::Value>,
) -> impl std::future::Future<Output = Result<(), TestBinaryError>> + Send {
async move {
let url = format!("{}/v1/namespaces/{namespace}/tables", self.catalog_url());
- let body = serde_json::json!({
+ let mut body = serde_json::json!({
"name": table,
"schema": {
"type": "struct",
@@ -292,6 +293,9 @@ pub trait IcebergOps: Sync {
]
}
});
+ if let Some(partition_spec) = partition_spec {
+ body["partition-spec"] = partition_spec;
+ }
let response = self
.http_client()
@@ -431,6 +435,8 @@ pub struct SnapshotSummary {
pub added_data_files: Option<String>,
#[serde(rename = "added-records")]
pub added_records: Option<String>,
+ #[serde(rename = "changed-partition-count")]
+ pub changed_partition_count: Option<String>,
}
impl IcebergOps for IcebergFixture {
@@ -502,6 +508,8 @@ fn create_http_client() -> HttpClient {
pub const DEFAULT_NAMESPACE: &str = "test";
pub const DEFAULT_TABLE: &str = "messages";
+/// Field id of the `active` column in the schema created by `IcebergOps::create_table`.
+const ACTIVE_FIELD_ID: i32 = 5;
pub struct IcebergPreCreatedFixture {
inner: IcebergFixture,
@@ -524,13 +532,61 @@ impl IcebergOps for IcebergPreCreatedFixture {
}
}
+impl IcebergPreCreatedFixture {
+ async fn setup_with_partition_spec(
+ partition_spec: Option<serde_json::Value>,
+ ) -> Result<Self, TestBinaryError> {
+ let inner = IcebergFixture::setup().await?;
+
+ inner.create_namespace(DEFAULT_NAMESPACE).await?;
+ inner
+ .create_table(DEFAULT_NAMESPACE, DEFAULT_TABLE, partition_spec)
+ .await?;
+
+ Ok(Self { inner })
+ }
+}
+
#[async_trait]
impl TestFixture for IcebergPreCreatedFixture {
async fn setup() -> Result<Self, TestBinaryError> {
- let inner = IcebergFixture::setup().await?;
+ Self::setup_with_partition_spec(None).await
+ }
- inner.create_namespace(DEFAULT_NAMESPACE).await?;
- inner.create_table(DEFAULT_NAMESPACE, DEFAULT_TABLE).await?;
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ self.inner.connectors_runtime_envs()
+ }
+}
+
+/// Same table as `IcebergPreCreatedFixture`, partitioned by identity(`active`).
+pub struct IcebergPartitionedTableFixture {
+ inner: IcebergPreCreatedFixture,
+}
+
+impl IcebergOps for IcebergPartitionedTableFixture {
+ fn catalog_url(&self) -> &str {
+ self.inner.catalog_url()
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ self.inner.http_client()
+ }
+}
+
+#[async_trait]
+impl TestFixture for IcebergPartitionedTableFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let partition_spec = serde_json::json!({
+ "spec-id": 0,
+ "fields": [{
+ "source-id": ACTIVE_FIELD_ID,
+ "field-id": 1000,
+ "name": "active",
+ "transform": "identity"
+ }]
+ });
+ let inner =
+ IcebergPreCreatedFixture::setup_with_partition_spec(Some(partition_spec)).await?;
Ok(Self { inner })
}
diff --git a/core/integration/tests/connectors/fixtures/iceberg/mod.rs b/core/integration/tests/connectors/fixtures/iceberg/mod.rs
index 670c2513c..0b5dd1877 100644
--- a/core/integration/tests/connectors/fixtures/iceberg/mod.rs
+++ b/core/integration/tests/connectors/fixtures/iceberg/mod.rs
@@ -18,5 +18,6 @@
mod container;
pub use container::{
- DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps, IcebergPreCreatedFixture,
+ DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps,
+ IcebergPartitionedTableFixture, IcebergPreCreatedFixture,
};
diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs
index 7eaf6fe51..e9d7a1332 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -63,7 +63,8 @@ pub use http::{
HttpSinkNdjsonFixture, HttpSinkNoMetadataFixture, HttpSinkRawFixture,
};
pub use iceberg::{
- DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps, IcebergPreCreatedFixture,
+ DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps,
+ IcebergPartitionedTableFixture, IcebergPreCreatedFixture,
};
pub use influxdb::{
InfluxDb3SinkFixture, InfluxDb3SourceFixture, InfluxDbSinkBase64Fixture, InfluxDbSinkFixture,
diff --git a/core/integration/tests/connectors/iceberg/iceberg_sink.rs b/core/integration/tests/connectors/iceberg/iceberg_sink.rs
index ceb082f6f..2e969775c 100644
--- a/core/integration/tests/connectors/iceberg/iceberg_sink.rs
+++ b/core/integration/tests/connectors/iceberg/iceberg_sink.rs
@@ -17,7 +17,8 @@
use crate::connectors::create_test_messages;
use crate::connectors::fixtures::{
- DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps, IcebergPreCreatedFixture,
+ DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps,
+ IcebergPartitionedTableFixture, IcebergPreCreatedFixture,
};
use bytes::Bytes;
use iggy::prelude::{IggyMessage, Partitioning};
@@ -33,6 +34,9 @@ const ICEBERG_SINK_KEY: &str = "iceberg";
const SNAPSHOT_POLL_ATTEMPTS: usize = 30;
const SNAPSHOT_POLL_INTERVAL_MS: u64 = 500;
const BULK_SNAPSHOT_POLL_ATTEMPTS: usize = 60;
+/// `create_test_messages` alternates `active`, so this many messages span both partition values.
+const PARTITIONED_MESSAGE_COUNT: usize = 5;
+const ACTIVE_PARTITION_COUNT: usize = 2;
#[iggy_harness(
server(connectors_runtime(config_path = "tests/connectors/iceberg/sink.toml")),
@@ -206,6 +210,82 @@ async fn iceberg_sink_handles_bulk_messages(
assert!(sinks[0].last_error.is_none());
}
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/iceberg/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn iceberg_sink_routes_messages_to_table_partitions(
+ harness: &TestHarness,
+ fixture: IcebergPartitionedTableFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+ let test_messages = create_test_messages(PARTITIONED_MESSAGE_COUNT);
+ let mut messages: Vec<IggyMessage> = test_messages
+ .iter()
+ .enumerate()
+ .map(|(i, msg)| {
+ let payload = serde_json::to_vec(msg).expect("Failed to serialize message");
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(Bytes::from(payload))
+ .build()
+ .expect("Failed to build message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("Failed to send messages");
+
+ fixture
+ .wait_for_snapshots(
+ DEFAULT_NAMESPACE,
+ DEFAULT_TABLE,
+ 1,
+ SNAPSHOT_POLL_ATTEMPTS,
+ SNAPSHOT_POLL_INTERVAL_MS,
+ )
+ .await
+ .expect("Data should be written to Iceberg table");
+
+ let metadata = fixture
+ .get_table_metadata(DEFAULT_NAMESPACE, DEFAULT_TABLE)
+ .await
+ .expect("Failed to get table metadata");
+ let snapshots = metadata
+ .metadata
+ .snapshots
+ .expect("Table should have a snapshot");
+ assert_eq!(snapshots.len(), 1, "One batch must produce one snapshot");
+ let summary = snapshots[0]
+ .summary
+ .as_ref()
+ .expect("Snapshot should have a summary");
+
+ assert_eq!(
+ summary.added_records.as_deref(),
+ Some(PARTITIONED_MESSAGE_COUNT.to_string().as_str())
+ );
+ assert_eq!(
+ summary.changed_partition_count.as_deref(),
+ Some(ACTIVE_PARTITION_COUNT.to_string().as_str())
+ );
+ assert_eq!(
+ summary.added_data_files.as_deref(),
+ Some(ACTIVE_PARTITION_COUNT.to_string().as_str())
+ );
+}
+
#[iggy_harness(
server(connectors_runtime(config_path = "tests/connectors/iceberg/sink_default_credentials.toml")),
seed = seeds::connector_stream
(iggy) 03/03: chore(connectors): mark clickhouse sink as unpublishable
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/iceberg-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3593e95ee762ac7a34c4807d6e00a121d9b42294
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Sep 1 13:51:59 2026 +0200
chore(connectors): mark clickhouse sink as unpublishable
All other sink crates set publish = false; clickhouse was the only
one missing it.
---
core/connectors/sinks/clickhouse_sink/Cargo.toml | 1 +
1 file changed, 1 insertion(+)
diff --git a/core/connectors/sinks/clickhouse_sink/Cargo.toml b/core/connectors/sinks/clickhouse_sink/Cargo.toml
index 07cf33456..3f0d07b0e 100644
--- a/core/connectors/sinks/clickhouse_sink/Cargo.toml
+++ b/core/connectors/sinks/clickhouse_sink/Cargo.toml
@@ -27,6 +27,7 @@ homepage = "https://iggy.apache.org"
documentation = "https://iggy.apache.org/docs"
repository = "https://github.com/apache/iggy"
readme = "../../README.md"
+publish = false
[lib]
crate-type = ["cdylib", "lib"]