You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "Shekharrajak (via GitHub)" <gi...@apache.org> on 2025/12/28 18:41:24 UTC
[PR] Feature/fluss table integration [doris]
Shekharrajak opened a new pull request, #59432:
URL: https://github.com/apache/doris/pull/59432
### What problem does this PR solve?
Issue Number: close #59220
Related PR: None
Problem Summary:
This PR implements support for reading Apache Fluss table data in Apache Doris. Fluss is a streaming storage system that provides real-time data access capabilities. This integration allows Doris users to query Fluss tables as external tables,.
The implementation includes:
- FE (Frontend) catalog layer with Fluss Java SDK integration
- Schema loading and type conversion from Fluss to Doris types
- Query planning with FlussScanNode
- BE (Backend) reader implementation wrapping Parquet/ORC readers
- Comprehensive unit test coverage
### Release note
**Feature**: Add support for reading Apache Fluss external tables
Users can now create Fluss catalogs and query Fluss tables in Doris:
CREATE CATALOG fluss_catalog PROPERTIES (
'type' = 'fluss',
'fluss.coordinator.uri' = 'localhost:9123'
);
SWITCH fluss_catalog;
SELECT * FROM database.table_name;### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [x] Yes. <!-- Add document PR link here. eg: https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on PR #59432:
URL: https://github.com/apache/doris/pull/59432#issuecomment-3838368072
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on PR #59432:
URL: https://github.com/apache/doris/pull/59432#issuecomment-3895347002
Hi @Shekharrajak thanks for your PR. This is large feature, besides the issue I commented, I suggest the following steps:
1. Submit a PR only contains docker compose script for Fluss, under `docker/thirdparties/`. This docker compose only provide a full-managed fluss env, for e2e test in future.
2. Submt another PR including the core code of Fluss Catalog/Database/Table/ScanNode, etc.
After these, I can help to add more e2e tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/fluss table integration [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #59432:
URL: https://github.com/apache/doris/pull/59432#issuecomment-3694952619
Thank you for your contribution to Apache Doris.
Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR).
Please clearly describe your PR:
1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on PR #59432:
URL: https://github.com/apache/doris/pull/59432#issuecomment-4544067385
>Submit a PR only contains docker compose script for Fluss, under docker/thirdparties/. This docker compose only provide a full-managed fluss env, for e2e test in future.
Drafting a PR in different branch
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #59432:
URL: https://github.com/apache/doris/pull/59432#discussion_r2757112136
##########
be/src/vec/exec/format/table/fluss_reader.cpp:
##########
@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "vec/exec/format/table/fluss_reader.h"
+
+#include "common/logging.h"
+#include "common/status.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+
+FlussReader::FlussReader(std::unique_ptr<GenericReader> file_format_reader,
+ RuntimeProfile* profile, RuntimeState* state,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ io::IOContext* io_ctx, FileMetaCache* meta_cache)
+ : TableFormatReader(std::move(file_format_reader), state, profile, params, range, io_ctx,
+ meta_cache) {
+ // Log tier information for debugging
+ if (range.__isset.table_format_params &&
+ range.table_format_params.__isset.fluss_params) {
+ const auto& fluss_params = range.table_format_params.fluss_params;
+ LOG(INFO) << "FlussReader initialized for table: "
+ << fluss_params.database_name << "." << fluss_params.table_name
+ << ", bucket: " << fluss_params.bucket_id
+ << ", format: " << fluss_params.file_format
+ << ", lake_snapshot_id: " << fluss_params.lake_snapshot_id
+ << ", lake_files: " << (fluss_params.__isset.lake_file_paths ?
+ fluss_params.lake_file_paths.size() : 0);
+ }
+}
+
+Status FlussReader::get_next_block_inner(Block* block, size_t* read_rows, bool* eof) {
+ // For MVP, we read from lake tier (Parquet/ORC files) using the underlying reader.
+ // The FE has already determined which files to read based on the LakeSnapshot.
+ // Future phases will add support for LOG_ONLY and HYBRID tiers via JNI bridge.
Review Comment:
We can also think of phasewise rollout :
Phase 1: MVP (Current)
Doris BE → Parquet/ORC Reader → Lake Tier Files (S3/MinIO)
- Implemented
- Only LAKE_ONLY tier support
Phase 2: JNI Bridge (Planned)
Doris BE -> JNI -> Java Fluss SDK -> Fluss TabletServer -> Log Tier Data
- Will add LOG_ONLY and HYBRID tier support
- Uses production Java SDK
- Follows established Doris pattern
Phase 3: Native Optimization (Future)
Doris BE → fluss-cpp/Rust bindings → Fluss TabletServer
- If/when C++ client reaches feature parity
- Performance optimization path
- Requires mature C++ client with:
- Full feature support
- Production stability
- Active maintenance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #59432:
URL: https://github.com/apache/doris/pull/59432#discussion_r2757109078
##########
be/src/vec/exec/format/table/fluss_reader.cpp:
##########
@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "vec/exec/format/table/fluss_reader.h"
+
+#include "common/logging.h"
+#include "common/status.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+
+FlussReader::FlussReader(std::unique_ptr<GenericReader> file_format_reader,
+ RuntimeProfile* profile, RuntimeState* state,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ io::IOContext* io_ctx, FileMetaCache* meta_cache)
+ : TableFormatReader(std::move(file_format_reader), state, profile, params, range, io_ctx,
+ meta_cache) {
+ // Log tier information for debugging
+ if (range.__isset.table_format_params &&
+ range.table_format_params.__isset.fluss_params) {
+ const auto& fluss_params = range.table_format_params.fluss_params;
+ LOG(INFO) << "FlussReader initialized for table: "
+ << fluss_params.database_name << "." << fluss_params.table_name
+ << ", bucket: " << fluss_params.bucket_id
+ << ", format: " << fluss_params.file_format
+ << ", lake_snapshot_id: " << fluss_params.lake_snapshot_id
+ << ", lake_files: " << (fluss_params.__isset.lake_file_paths ?
+ fluss_params.lake_file_paths.size() : 0);
+ }
+}
+
+Status FlussReader::get_next_block_inner(Block* block, size_t* read_rows, bool* eof) {
+ // For MVP, we read from lake tier (Parquet/ORC files) using the underlying reader.
+ // The FE has already determined which files to read based on the LakeSnapshot.
+ // Future phases will add support for LOG_ONLY and HYBRID tiers via JNI bridge.
Review Comment:
Fluss C++ client has LIMITED features compared to the Java client but we can brainstorm around this and come up with best long term approach. Adding fluss-cpp dependency should be okay, if we really get best performance & usuability.
Doris already uses JNI for similar integrations: HudiJniReader, PaimonJniReader , hence I was planning around FlussJniReader.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #59432:
URL: https://github.com/apache/doris/pull/59432#discussion_r2802596812
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/fluss/FlussExternalCatalogFactory.java:
##########
@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.fluss;
+
+import org.apache.doris.common.DdlException;
+import org.apache.doris.datasource.ExternalCatalog;
+
+import java.util.Map;
+
+public class FlussExternalCatalogFactory {
Review Comment:
I think this factory is unnecessary?
##########
.github/workflows/fluss-integration.yml:
##########
@@ -0,0 +1,270 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Fluss Integration Tests
Review Comment:
We don't run e2e test like this.
You can refer to `docker/thirdparties/`. see how we test with Paimon.
Shortly speaking, you need to build a docker env only for fluss, and run e2e test written in `regression-test/suites/external_table_p0/`
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/fluss/FlussMetadataOps.java:
##########
@@ -0,0 +1,495 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.fluss;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.config.Configuration;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
+
+public class FlussMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(FlussMetadataOps.class);
+
+ private static final int MAX_RETRY_ATTEMPTS = 3;
+ private static final long INITIAL_RETRY_DELAY_MS = 100;
+ private static final long MAX_RETRY_DELAY_MS = 5000;
+
+ private final FlussExternalCatalog catalog;
+ private final String bootstrapServers;
+
+ private final Map<String, FlussExternalTable.FlussTableMetadata> tableMetadataCache;
+ private final Map<String, List<String>> databaseTablesCache;
+ private final ReadWriteLock cacheLock = new ReentrantReadWriteLock();
+
+ private volatile Connection connection;
+ private volatile Admin admin;
+ private final AtomicBoolean initialized = new AtomicBoolean(false);
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final Object connectionLock = new Object();
+
+ public FlussMetadataOps(FlussExternalCatalog catalog) {
Review Comment:
This constructor is not used
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/fluss/FlussExternalCatalog.java:
##########
@@ -0,0 +1,232 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.fluss;
+
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.CatalogProperty;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.InitCatalogLog;
+import org.apache.doris.datasource.SessionContext;
+import org.apache.doris.datasource.operations.ExternalMetadataOperations;
+import org.apache.doris.transaction.TransactionManagerFactory;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.TableNotExistException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public class FlussExternalCatalog extends ExternalCatalog {
+ private static final Logger LOG = LogManager.getLogger(FlussExternalCatalog.class);
+
+ public static final String FLUSS_COORDINATOR_URI = "fluss.coordinator.uri";
+ public static final String FLUSS_BOOTSTRAP_SERVERS = "bootstrap.servers";
+ public static final String FLUSS_SECURITY_PROTOCOL = "fluss.security.protocol";
+ public static final String FLUSS_SASL_MECHANISM = "fluss.sasl.mechanism";
+ public static final String FLUSS_SASL_USERNAME = "fluss.sasl.username";
+ public static final String FLUSS_SASL_PASSWORD = "fluss.sasl.password";
+ public static final String FLUSS_ENABLE_MAPPING_VARBINARY = "fluss.enable.mapping.varbinary";
+ public static final String FLUSS_TABLE_META_CACHE_TTL_SECOND = "fluss.table.meta.cache.ttl.second";
+
+ protected Connection flussConnection;
+ protected Admin flussAdmin;
+
+ public FlussExternalCatalog(long catalogId, String name, String resource, Map<String, String> props,
+ String comment) {
+ super(catalogId, name, InitCatalogLog.Type.FLUSS, comment);
+ this.catalogProperty = new CatalogProperty(resource, props);
+ }
+
+ @Override
+ public void checkProperties() throws DdlException {
+ super.checkProperties();
+ String coordinatorUri = catalogProperty.getOrDefault(FLUSS_COORDINATOR_URI, null);
+ String bootstrapServers = catalogProperty.getOrDefault(FLUSS_BOOTSTRAP_SERVERS, null);
+ if (StringUtils.isEmpty(coordinatorUri) && StringUtils.isEmpty(bootstrapServers)) {
+ throw new DdlException("Missing required property: " + FLUSS_COORDINATOR_URI
+ + " or " + FLUSS_BOOTSTRAP_SERVERS);
+ }
+ }
+
+ @Override
+ protected void initLocalObjectsImpl() {
+ Configuration conf = createFlussConfiguration();
+ flussConnection = ConnectionFactory.createConnection(conf);
+ flussAdmin = flussConnection.getAdmin();
+ initPreExecutionAuthenticator();
+ FlussMetadataOps ops = ExternalMetadataOperations.newFlussMetadataOps(this, flussConnection);
+ threadPoolWithPreAuth = ThreadPoolManager.newDaemonFixedThreadPoolWithPreAuth(
Review Comment:
No need to init this thread pool, in case you really use it in Fluss Catalog
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/fluss/FlussMetadataOps.java:
##########
@@ -0,0 +1,495 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.fluss;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.config.Configuration;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
+
+public class FlussMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(FlussMetadataOps.class);
+
+ private static final int MAX_RETRY_ATTEMPTS = 3;
+ private static final long INITIAL_RETRY_DELAY_MS = 100;
+ private static final long MAX_RETRY_DELAY_MS = 5000;
+
+ private final FlussExternalCatalog catalog;
+ private final String bootstrapServers;
+
+ private final Map<String, FlussExternalTable.FlussTableMetadata> tableMetadataCache;
+ private final Map<String, List<String>> databaseTablesCache;
+ private final ReadWriteLock cacheLock = new ReentrantReadWriteLock();
+
+ private volatile Connection connection;
+ private volatile Admin admin;
+ private final AtomicBoolean initialized = new AtomicBoolean(false);
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final Object connectionLock = new Object();
+
+ public FlussMetadataOps(FlussExternalCatalog catalog) {
+ this.catalog = catalog;
+ this.bootstrapServers = catalog.getBootstrapServers();
+ this.tableMetadataCache = new HashMap<>();
+ this.databaseTablesCache = new HashMap<>();
+ }
+
+ public FlussMetadataOps(org.apache.doris.datasource.ExternalCatalog catalog, Connection connection) {
Review Comment:
```suggestion
public FlussMetadataOps(FlussExternalCatalog catalog, Connection connection) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/fluss/FlussExternalCatalog.java:
##########
@@ -0,0 +1,232 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.fluss;
+
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.CatalogProperty;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.InitCatalogLog;
+import org.apache.doris.datasource.SessionContext;
+import org.apache.doris.datasource.operations.ExternalMetadataOperations;
+import org.apache.doris.transaction.TransactionManagerFactory;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.TableNotExistException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public class FlussExternalCatalog extends ExternalCatalog {
+ private static final Logger LOG = LogManager.getLogger(FlussExternalCatalog.class);
+
+ public static final String FLUSS_COORDINATOR_URI = "fluss.coordinator.uri";
Review Comment:
You need to define a `FlussProperties` class, extends from `MetastoreProperties`.
And all parsing logic should in `FlussProperties`
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/fluss/FlussMetadataOps.java:
##########
@@ -0,0 +1,495 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.fluss;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.config.Configuration;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
+
+public class FlussMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(FlussMetadataOps.class);
+
+ private static final int MAX_RETRY_ATTEMPTS = 3;
+ private static final long INITIAL_RETRY_DELAY_MS = 100;
+ private static final long MAX_RETRY_DELAY_MS = 5000;
+
+ private final FlussExternalCatalog catalog;
+ private final String bootstrapServers;
+
+ private final Map<String, FlussExternalTable.FlussTableMetadata> tableMetadataCache;
Review Comment:
You can refer to `fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractIcebergProperties.java` to see how cache framework work.
Or maybe we can simply remove this cache in this MVP version, and add it later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #59432:
URL: https://github.com/apache/doris/pull/59432#discussion_r2802707887
##########
be/src/vec/exec/format/table/fluss_reader.cpp:
##########
@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "vec/exec/format/table/fluss_reader.h"
+
+#include "common/logging.h"
+#include "common/status.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+
+FlussReader::FlussReader(std::unique_ptr<GenericReader> file_format_reader,
+ RuntimeProfile* profile, RuntimeState* state,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ io::IOContext* io_ctx, FileMetaCache* meta_cache)
+ : TableFormatReader(std::move(file_format_reader), state, profile, params, range, io_ctx,
+ meta_cache) {
+ // Log tier information for debugging
+ if (range.__isset.table_format_params &&
+ range.table_format_params.__isset.fluss_params) {
+ const auto& fluss_params = range.table_format_params.fluss_params;
+ LOG(INFO) << "FlussReader initialized for table: "
+ << fluss_params.database_name << "." << fluss_params.table_name
+ << ", bucket: " << fluss_params.bucket_id
+ << ", format: " << fluss_params.file_format
+ << ", lake_snapshot_id: " << fluss_params.lake_snapshot_id
+ << ", lake_files: " << (fluss_params.__isset.lake_file_paths ?
+ fluss_params.lake_file_paths.size() : 0);
+ }
+}
+
+Status FlussReader::get_next_block_inner(Block* block, size_t* read_rows, bool* eof) {
+ // For MVP, we read from lake tier (Parquet/ORC files) using the underlying reader.
+ // The FE has already determined which files to read based on the LakeSnapshot.
+ // Future phases will add support for LOG_ONLY and HYBRID tiers via JNI bridge.
Review Comment:
LGTM!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #59432:
URL: https://github.com/apache/doris/pull/59432#issuecomment-3838576096
# Cloud UT Coverage Report
Increment line coverage ` ` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/0775573352dc973fab6045e91c40b9aaab524e46_0775573352dc973fab6045e91c40b9aaab524e46_cloud/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/0775573352dc973fab6045e91c40b9aaab524e46_0775573352dc973fab6045e91c40b9aaab524e46_cloud/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 79.29% (1792/2260) |
| Line Coverage | 64.79% (31849/49158) |
| Region Coverage | 65.46% (15894/24280) |
| Branch Coverage | 56.00% (8444/15078) |
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on code in PR #59432:
URL: https://github.com/apache/doris/pull/59432#discussion_r2757112136
##########
be/src/vec/exec/format/table/fluss_reader.cpp:
##########
@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "vec/exec/format/table/fluss_reader.h"
+
+#include "common/logging.h"
+#include "common/status.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+
+FlussReader::FlussReader(std::unique_ptr<GenericReader> file_format_reader,
+ RuntimeProfile* profile, RuntimeState* state,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ io::IOContext* io_ctx, FileMetaCache* meta_cache)
+ : TableFormatReader(std::move(file_format_reader), state, profile, params, range, io_ctx,
+ meta_cache) {
+ // Log tier information for debugging
+ if (range.__isset.table_format_params &&
+ range.table_format_params.__isset.fluss_params) {
+ const auto& fluss_params = range.table_format_params.fluss_params;
+ LOG(INFO) << "FlussReader initialized for table: "
+ << fluss_params.database_name << "." << fluss_params.table_name
+ << ", bucket: " << fluss_params.bucket_id
+ << ", format: " << fluss_params.file_format
+ << ", lake_snapshot_id: " << fluss_params.lake_snapshot_id
+ << ", lake_files: " << (fluss_params.__isset.lake_file_paths ?
+ fluss_params.lake_file_paths.size() : 0);
+ }
+}
+
+Status FlussReader::get_next_block_inner(Block* block, size_t* read_rows, bool* eof) {
+ // For MVP, we read from lake tier (Parquet/ORC files) using the underlying reader.
+ // The FE has already determined which files to read based on the LakeSnapshot.
+ // Future phases will add support for LOG_ONLY and HYBRID tiers via JNI bridge.
Review Comment:
We can also think of phasewise rollout :
Phase 1: MVP (Current)
Doris BE → Parquet/ORC Reader → Lake Tier Files (S3/MinIO)
- Implemented
- Only LAKE_ONLY tier support
Phase 2: JNI Bridge (Planned)
Doris BE -> JNI -> Java Fluss SDK -> Fluss TabletServer -> Log Tier Data
- Will add LOG_ONLY and HYBRID tier support
- Uses Java SDK - since that have more stable features
Phase 3: Native Optimization (Future)
Doris BE → fluss-cpp/Rust bindings → Fluss TabletServer
- If/when C++ client reaches feature parity
- Performance optimization path
- Collaborate with fluss team to make cpp client with all the features required that we have for java SDK already implemented
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on PR #59432:
URL: https://github.com/apache/doris/pull/59432#issuecomment-3855716363
Please help in validating
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "Shekharrajak (via GitHub)" <gi...@apache.org>.
Shekharrajak commented on PR #59432:
URL: https://github.com/apache/doris/pull/59432#issuecomment-3834146987
Please trigger the workflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](fluss) fluss table integration [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #59432:
URL: https://github.com/apache/doris/pull/59432#discussion_r2756785888
##########
be/src/vec/exec/format/table/fluss_reader.cpp:
##########
@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "vec/exec/format/table/fluss_reader.h"
+
+#include "common/logging.h"
+#include "common/status.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+
+FlussReader::FlussReader(std::unique_ptr<GenericReader> file_format_reader,
+ RuntimeProfile* profile, RuntimeState* state,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ io::IOContext* io_ctx, FileMetaCache* meta_cache)
+ : TableFormatReader(std::move(file_format_reader), state, profile, params, range, io_ctx,
+ meta_cache) {
+ // Log tier information for debugging
+ if (range.__isset.table_format_params &&
+ range.table_format_params.__isset.fluss_params) {
+ const auto& fluss_params = range.table_format_params.fluss_params;
+ LOG(INFO) << "FlussReader initialized for table: "
+ << fluss_params.database_name << "." << fluss_params.table_name
+ << ", bucket: " << fluss_params.bucket_id
+ << ", format: " << fluss_params.file_format
+ << ", lake_snapshot_id: " << fluss_params.lake_snapshot_id
+ << ", lake_files: " << (fluss_params.__isset.lake_file_paths ?
+ fluss_params.lake_file_paths.size() : 0);
+ }
+}
+
+Status FlussReader::get_next_block_inner(Block* block, size_t* read_rows, bool* eof) {
+ // For MVP, we read from lake tier (Parquet/ORC files) using the underlying reader.
+ // The FE has already determined which files to read based on the LakeSnapshot.
+ // Future phases will add support for LOG_ONLY and HYBRID tiers via JNI bridge.
Review Comment:
Why do we still need JNI bridge? Since we already bring `fluss-cpp` in?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org