diff --git a/be/src/olap/push_handler.cpp b/be/src/olap/push_handler.cpp index 1a92f0382921540..04d2910f82e0225 100644 --- a/be/src/olap/push_handler.cpp +++ b/be/src/olap/push_handler.cpp @@ -323,7 +323,6 @@ PushBrokerReader::PushBrokerReader(const Schema* schema, const TBrokerScanRange& if (0 == _ranges.size()) { return; } - _file_params.file_type = _ranges[0].file_type; _file_params.format_type = _ranges[0].format_type; _file_params.src_tuple_id = _params.src_tuple_id; _file_params.dest_tuple_id = _params.dest_tuple_id; @@ -337,6 +336,12 @@ PushBrokerReader::PushBrokerReader(const Schema* schema, const TBrokerScanRange& for (int i = 0; i < _ranges.size(); ++i) { TFileRangeDesc file_range; + // TODO(cmy): in previous implementation, the file_type is set in _file_params + // and it use _ranges[0].file_type. + // Later, this field is moved to TFileRangeDesc, but here we still only use _ranges[0]'s + // file_type. + // Because I don't know if othere range has this field, so just keep it same as before. + file_range.file_type = _ranges[0].file_type; file_range.load_id = _ranges[i].load_id; file_range.path = _ranges[i].path; file_range.start_offset = _ranges[i].start_offset; diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp index cea59acdfb3c792..87110ac7dec83b3 100644 --- a/be/src/runtime/fragment_mgr.cpp +++ b/be/src/runtime/fragment_mgr.cpp @@ -672,6 +672,13 @@ Status FragmentMgr::_get_query_ctx(const Params& params, TUniqueId query_id, boo query_ctx->query_id = query_id; RETURN_IF_ERROR(DescriptorTbl::create(&(query_ctx->obj_pool), params.desc_tbl, &(query_ctx->desc_tbl))); + + // set file scan range params + if (params.__isset.file_scan_params) { + LOG(INFO) << "yy debug set file_scan_range_params_map size: " << params.file_scan_params.size(); + query_ctx->file_scan_range_params_map = params.file_scan_params; + } + query_ctx->coord_addr = params.coord; LOG(INFO) << "query_id: " << UniqueId(query_ctx->query_id.hi, query_ctx->query_id.lo) << " coord_addr " << query_ctx->coord_addr @@ -762,7 +769,7 @@ Status FragmentMgr::exec_plan_fragment(const TExecPlanFragmentParams& params, cur_span->SetAttribute("query_id", print_id(params.params.query_id)); cur_span->SetAttribute("instance_id", print_id(params.params.fragment_instance_id)); - VLOG_ROW << "exec_plan_fragment params is " + LOG(INFO) << "exec_plan_fragment params is " << apache::thrift::ThriftDebugString(params).c_str(); // sometimes TExecPlanFragmentParams debug string is too long and glog // will truncate the log line, so print query options seperately for debuggin purpose diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h index 8ed0150db5a40f3..c7d57629a2a1796 100644 --- a/be/src/runtime/query_context.h +++ b/be/src/runtime/query_context.h @@ -197,6 +197,8 @@ class QueryContext { std::vector fragment_ids; + std::map file_scan_range_params_map; + private: ExecEnv* _exec_env; vectorized::VecDateTimeValue _start_time; diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp index 7a936487c4d2e05..9e50b56689a664b 100644 --- a/be/src/service/internal_service.cpp +++ b/be/src/service/internal_service.cpp @@ -444,6 +444,7 @@ void PInternalServiceImpl::tablet_writer_cancel(google::protobuf::RpcController* Status PInternalServiceImpl::_exec_plan_fragment_impl(const std::string& ser_request, PFragmentRequestVersion version, bool compact) { + LOG(INFO) << "yy debug _exec_plan_fragment_impl ser_request size: " << ser_request.size(); // Sometimes the BE do not receive the first heartbeat message and it receives request from FE // If BE execute this fragment, it will core when it wants to get some property from master info. if (ExecEnv::GetInstance()->master_info() == nullptr) { diff --git a/be/src/vec/exec/format/csv/csv_reader.cpp b/be/src/vec/exec/format/csv/csv_reader.cpp index 8db857940206cd3..c72419f8a3042b5 100644 --- a/be/src/vec/exec/format/csv/csv_reader.cpp +++ b/be/src/vec/exec/format/csv/csv_reader.cpp @@ -88,7 +88,12 @@ CsvReader::CsvReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounte _io_ctx(io_ctx) { _file_format_type = _params.format_type; _is_proto_format = _file_format_type == TFileFormatType::FORMAT_PROTO; - _file_compress_type = _params.compress_type; + if (_range.__isset.compress_type) { + // for compatibility + _file_compress_type = _range.compress_type; + } else { + _file_compress_type = _params.compress_type; + } _size = _range.size; _text_converter.reset(new (std::nothrow) TextConverter('\\')); @@ -111,7 +116,12 @@ CsvReader::CsvReader(RuntimeProfile* profile, const TFileScanRangeParams& params _decompressor(nullptr), _io_ctx(io_ctx) { _file_format_type = _params.format_type; - _file_compress_type = _params.compress_type; + if (_range.__isset.compress_type) { + // for compatibility + _file_compress_type = _range.compress_type; + } else { + _file_compress_type = _params.compress_type; + } _size = _range.size; _init_system_properties(); _init_file_description(); @@ -120,7 +130,12 @@ CsvReader::CsvReader(RuntimeProfile* profile, const TFileScanRangeParams& params CsvReader::~CsvReader() = default; void CsvReader::_init_system_properties() { - _system_properties.system_type = _params.file_type; + if (_range.__isset.file_type) { + // for compatibility + _system_properties.system_type = _range.file_type; + } else { + _system_properties.system_type = _params.file_type; + } _system_properties.properties = _params.properties; _system_properties.hdfs_params = _params.hdfs_params; if (_params.__isset.broker_addresses) { diff --git a/be/src/vec/exec/format/json/new_json_reader.cpp b/be/src/vec/exec/format/json/new_json_reader.cpp index f6eabaa7cd43750..7ad7f178a3e0882 100644 --- a/be/src/vec/exec/format/json/new_json_reader.cpp +++ b/be/src/vec/exec/format/json/new_json_reader.cpp @@ -135,7 +135,12 @@ NewJsonReader::NewJsonReader(RuntimeProfile* profile, const TFileScanRangeParams } void NewJsonReader::_init_system_properties() { - _system_properties.system_type = _params.file_type; + if (_range.__isset.file_type) { + // for compatibility + _system_properties.system_type = _range.file_type; + } else { + _system_properties.system_type = _params.file_type; + } _system_properties.properties = _params.properties; _system_properties.hdfs_params = _params.hdfs_params; if (_params.__isset.broker_addresses) { diff --git a/be/src/vec/exec/format/orc/vorc_reader.cpp b/be/src/vec/exec/format/orc/vorc_reader.cpp index 7f87f77590d553b..4ef5f4a8cb25e70 100644 --- a/be/src/vec/exec/format/orc/vorc_reader.cpp +++ b/be/src/vec/exec/format/orc/vorc_reader.cpp @@ -864,7 +864,12 @@ void OrcReader::_init_bloom_filter( } void OrcReader::_init_system_properties() { - _system_properties.system_type = _scan_params.file_type; + if (_scan_range.__isset.file_type) { + // for compatibility + _system_properties.system_type = _scan_range.file_type; + } else { + _system_properties.system_type = _scan_params.file_type; + } _system_properties.properties = _scan_params.properties; _system_properties.hdfs_params = _scan_params.hdfs_params; if (_scan_params.__isset.broker_addresses) { diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp index 9b179384e25f198..58f8a52c24c12b9 100644 --- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp +++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp @@ -269,7 +269,12 @@ Status ParquetReader::open() { } void ParquetReader::_init_system_properties() { - _system_properties.system_type = _scan_params.file_type; + if (_scan_range.__isset.file_type) { + // for compatibility + _system_properties.system_type = _scan_range.file_type; + } else { + _system_properties.system_type = _scan_params.file_type; + } _system_properties.properties = _scan_params.properties; _system_properties.hdfs_params = _scan_params.hdfs_params; if (_scan_params.__isset.broker_addresses) { diff --git a/be/src/vec/exec/scan/avro_jni_reader.cpp b/be/src/vec/exec/scan/avro_jni_reader.cpp index 5d1ef40cbc87aca..ffca8682c488009 100644 --- a/be/src/vec/exec/scan/avro_jni_reader.cpp +++ b/be/src/vec/exec/scan/avro_jni_reader.cpp @@ -74,7 +74,13 @@ Status AvroJNIReader::init_fetch_table_reader( index++; } - TFileType::type type = _params.file_type; + TFileType::type type; + if (_range.__isset.file_type) { + // for compatibility + type = _range.file_type; + } else { + type = _params.file_type; + } std::map required_param = { {"required_fields", required_fields.str()}, {"columns_types", columns_types.str()}, diff --git a/be/src/vec/exec/scan/new_file_scan_node.cpp b/be/src/vec/exec/scan/new_file_scan_node.cpp index 127539d26e4b14c..d327cf10426ef5a 100644 --- a/be/src/vec/exec/scan/new_file_scan_node.cpp +++ b/be/src/vec/exec/scan/new_file_scan_node.cpp @@ -49,6 +49,12 @@ Status NewFileScanNode::init(const TPlanNode& tnode, RuntimeState* state) { Status NewFileScanNode::prepare(RuntimeState* state) { RETURN_IF_ERROR(VScanNode::prepare(state)); + if (state->get_query_ctx() != nullptr + && state->get_query_ctx()->file_scan_range_params_map.count(id()) > 0) { + TFileScanRangeParams& params = state->get_query_ctx()->file_scan_range_params_map[id()]; + _input_tuple_id = params.src_tuple_id; + _output_tuple_id = params.dest_tuple_id; + } return Status::OK(); } @@ -74,8 +80,10 @@ void NewFileScanNode::set_scan_ranges(const std::vector& scan_ _scan_ranges.shrink_to_fit(); LOG(INFO) << "Merge " << scan_ranges.size() << " scan ranges to " << _scan_ranges.size(); } - if (scan_ranges.size() > 0) { - _input_tuple_id = + if (scan_ranges.size() > 0 && scan_ranges[0].scan_range.ext_scan_range.file_scan_range.__isset.params) { + // for compatibility. + // in new implement, the tuple id is set in prepare phase + _input_tuple_id = scan_ranges[0].scan_range.ext_scan_range.file_scan_range.params.src_tuple_id; _output_tuple_id = scan_ranges[0].scan_range.ext_scan_range.file_scan_range.params.dest_tuple_id; diff --git a/be/src/vec/exec/scan/vfile_scanner.cpp b/be/src/vec/exec/scan/vfile_scanner.cpp index b95d2aa773b0645..01b0765564e6085 100644 --- a/be/src/vec/exec/scan/vfile_scanner.cpp +++ b/be/src/vec/exec/scan/vfile_scanner.cpp @@ -89,7 +89,6 @@ VFileScanner::VFileScanner(RuntimeState* state, NewFileScanNode* parent, int64_t const TFileScanRange& scan_range, RuntimeProfile* profile, ShardedKVCache* kv_cache) : VScanner(state, static_cast(parent), limit, profile), - _params(scan_range.params), _ranges(scan_range.ranges), _next_range(0), _cur_reader(nullptr), @@ -99,6 +98,18 @@ VFileScanner::VFileScanner(RuntimeState* state, NewFileScanNode* parent, int64_t if (scan_range.params.__isset.strict_mode) { _strict_mode = scan_range.params.strict_mode; } + + LOG(INFO) << "yy debug is null: " << (state->get_query_ctx() != nullptr) + << ", find: " << state->get_query_ctx()->file_scan_range_params_map.count(parent->id()) + << ", id: " << parent->id(); + if (state->get_query_ctx() != nullptr + && state->get_query_ctx()->file_scan_range_params_map.count(parent->id()) > 0) { + LOG(INFO) << "yy debug got from query context"; + _params = &(state->get_query_ctx()->file_scan_range_params_map[parent->id()]); + } else { + CHECK(scan_range.__isset.params); + _params = &(scan_range.params); + } } Status VFileScanner::prepare( @@ -133,13 +144,13 @@ Status VFileScanner::prepare( std::vector({_input_tuple_desc->id()}), std::vector({false}))); // prepare pre filters - if (_params.__isset.pre_filter_exprs_list) { + if (_params->__isset.pre_filter_exprs_list) { RETURN_IF_ERROR(doris::vectorized::VExpr::create_expr_trees( - _params.pre_filter_exprs_list, _pre_conjunct_ctxs)); - } else if (_params.__isset.pre_filter_exprs) { + _params->pre_filter_exprs_list, _pre_conjunct_ctxs)); + } else if (_params->__isset.pre_filter_exprs) { VExprContextSPtr context; RETURN_IF_ERROR( - doris::vectorized::VExpr::create_expr_tree(_params.pre_filter_exprs, context)); + doris::vectorized::VExpr::create_expr_tree(_params->pre_filter_exprs, context)); _pre_conjunct_ctxs.emplace_back(context); } @@ -569,9 +580,9 @@ Status VFileScanner::_get_next_reader() { // create reader for specific format Status init_status; - TFileFormatType::type format_type = _params.format_type; + TFileFormatType::type format_type = _params->format_type; // JNI reader can only push down column value range - bool push_down_predicates = !_is_load && _params.format_type != TFileFormatType::FORMAT_JNI; + bool push_down_predicates = !_is_load && _params->format_type != TFileFormatType::FORMAT_JNI; if (format_type == TFileFormatType::FORMAT_JNI && range.__isset.table_format_params && range.table_format_params.table_format_type == "hudi") { if (range.table_format_params.hudi_params.delta_logs.empty()) { @@ -599,7 +610,7 @@ Status VFileScanner::_get_next_reader() { } else if (range.__isset.table_format_params && range.table_format_params.table_format_type == "hudi") { _cur_reader = - HudiJniReader::create_unique(_params, range.table_format_params.hudi_params, + HudiJniReader::create_unique(*_params, range.table_format_params.hudi_params, _file_slot_descs, _state, _profile); init_status = ((HudiJniReader*)_cur_reader.get())->init_reader(_colname_to_value_range); @@ -608,7 +619,7 @@ Status VFileScanner::_get_next_reader() { } case TFileFormatType::FORMAT_PARQUET: { std::unique_ptr parquet_reader = ParquetReader::create_unique( - _profile, _params, range, _state->query_options().batch_size, + _profile, *_params, range, _state->query_options().batch_size, const_cast(&_state->timezone_obj()), _io_ctx.get(), _state, ExecEnv::GetInstance()->file_meta_cache(), _state->query_options().enable_parquet_lazy_mat); @@ -627,7 +638,7 @@ Status VFileScanner::_get_next_reader() { range.table_format_params.table_format_type == "iceberg") { std::unique_ptr iceberg_reader = IcebergTableReader::create_unique(std::move(parquet_reader), _profile, - _state, _params, range, _kv_cache, + _state, *_params, range, _kv_cache, _io_ctx.get()); init_status = iceberg_reader->init_reader( _file_col_names, _col_id_name_map, _colname_to_value_range, @@ -649,7 +660,7 @@ Status VFileScanner::_get_next_reader() { } case TFileFormatType::FORMAT_ORC: { std::unique_ptr orc_reader = OrcReader::create_unique( - _profile, _state, _params, range, _state->query_options().batch_size, + _profile, _state, *_params, range, _state->query_options().batch_size, _state->timezone(), _io_ctx.get(), _state->query_options().enable_orc_lazy_mat); if (push_down_predicates && _push_down_conjuncts.empty() && !_conjuncts.empty()) { _push_down_conjuncts.resize(_conjuncts.size()); @@ -662,7 +673,7 @@ Status VFileScanner::_get_next_reader() { range.table_format_params.table_format_type == "transactional_hive") { std::unique_ptr tran_orc_reader = TransactionalHiveReader::create_unique(std::move(orc_reader), _profile, - _state, _params, range, + _state, *_params, range, _io_ctx.get()); init_status = tran_orc_reader->init_reader( _file_col_names, _colname_to_value_range, _push_down_conjuncts, @@ -686,13 +697,13 @@ Status VFileScanner::_get_next_reader() { case TFileFormatType::FORMAT_CSV_LZOP: case TFileFormatType::FORMAT_CSV_DEFLATE: case TFileFormatType::FORMAT_PROTO: { - _cur_reader = CsvReader::create_unique(_state, _profile, &_counter, _params, range, + _cur_reader = CsvReader::create_unique(_state, _profile, &_counter, *_params, range, _file_slot_descs, _io_ctx.get()); init_status = ((CsvReader*)(_cur_reader.get()))->init_reader(_is_load); break; } case TFileFormatType::FORMAT_JSON: { - _cur_reader = NewJsonReader::create_unique(_state, _profile, &_counter, _params, range, + _cur_reader = NewJsonReader::create_unique(_state, _profile, &_counter, *_params, range, _file_slot_descs, &_scanner_eof, _io_ctx.get(), _is_dynamic_schema); init_status = @@ -700,13 +711,13 @@ Status VFileScanner::_get_next_reader() { break; } case TFileFormatType::FORMAT_AVRO: { - _cur_reader = AvroJNIReader::create_unique(_state, _profile, _params, _file_slot_descs); + _cur_reader = AvroJNIReader::create_unique(_state, _profile, *_params, _file_slot_descs); init_status = ((AvroJNIReader*)(_cur_reader.get())) ->init_fetch_table_reader(_colname_to_value_range); break; } default: - return Status::InternalError("Not supported file format: {}", _params.format_type); + return Status::InternalError("Not supported file format: {}", _params->format_type); } if (init_status.is()) { @@ -810,8 +821,8 @@ Status VFileScanner::_init_expr_ctxes() { } } - _num_of_columns_from_file = _params.num_of_columns_from_file; - for (const auto& slot_info : _params.required_slots) { + _num_of_columns_from_file = _params->num_of_columns_from_file; + for (const auto& slot_info : _params->required_slots) { auto slot_id = slot_info.slot_id; auto it = full_src_slot_map.find(slot_id); if (it == std::end(full_src_slot_map)) { @@ -843,8 +854,8 @@ Status VFileScanner::_init_expr_ctxes() { continue; } vectorized::VExprContextSPtr ctx; - auto it = _params.default_value_of_src_slot.find(slot_desc->id()); - if (it != std::end(_params.default_value_of_src_slot)) { + auto it = _params->default_value_of_src_slot.find(slot_desc->id()); + if (it != std::end(_params->default_value_of_src_slot)) { if (!it->second.nodes.empty()) { RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(it->second, ctx)); RETURN_IF_ERROR(ctx->prepare(_state, *_default_val_row_desc)); @@ -857,14 +868,14 @@ Status VFileScanner::_init_expr_ctxes() { if (_is_load) { // follow desc expr map is only for load task. - bool has_slot_id_map = _params.__isset.dest_sid_to_src_sid_without_trans; + bool has_slot_id_map = _params->__isset.dest_sid_to_src_sid_without_trans; int idx = 0; for (auto slot_desc : _output_tuple_desc->slots()) { if (!slot_desc->is_materialized()) { continue; } - auto it = _params.expr_of_dest_slot.find(slot_desc->id()); - if (it == std::end(_params.expr_of_dest_slot)) { + auto it = _params->expr_of_dest_slot.find(slot_desc->id()); + if (it == std::end(_params->expr_of_dest_slot)) { return Status::InternalError("No expr for dest slot, id={}, name={}", slot_desc->id(), slot_desc->col_name()); } @@ -879,8 +890,8 @@ Status VFileScanner::_init_expr_ctxes() { _dest_slot_name_to_idx[slot_desc->col_name()] = idx++; if (has_slot_id_map) { - auto it1 = _params.dest_sid_to_src_sid_without_trans.find(slot_desc->id()); - if (it1 == std::end(_params.dest_sid_to_src_sid_without_trans)) { + auto it1 = _params->dest_sid_to_src_sid_without_trans.find(slot_desc->id()); + if (it1 == std::end(_params->dest_sid_to_src_sid_without_trans)) { _src_slot_descs_order_by_dest.emplace_back(nullptr); } else { auto _src_slot_it = full_src_slot_map.find(it1->second); diff --git a/be/src/vec/exec/scan/vfile_scanner.h b/be/src/vec/exec/scan/vfile_scanner.h index 0b1d7a558a02481..9a16d056edba9d1 100644 --- a/be/src/vec/exec/scan/vfile_scanner.h +++ b/be/src/vec/exec/scan/vfile_scanner.h @@ -89,7 +89,7 @@ class VFileScanner : public VScanner { protected: std::unique_ptr _text_converter; - const TFileScanRangeParams& _params; + const TFileScanRangeParams* _params; const std::vector& _ranges; int _next_range; diff --git a/be/src/vec/exec/scan/vscanner.cpp b/be/src/vec/exec/scan/vscanner.cpp index 47c2896e2de6805..1298d3eb55a9179 100644 --- a/be/src/vec/exec/scan/vscanner.cpp +++ b/be/src/vec/exec/scan/vscanner.cpp @@ -38,6 +38,7 @@ VScanner::VScanner(RuntimeState* state, VScanNode* parent, int64_t limit, Runtim _real_tuple_desc = _input_tuple_desc != nullptr ? _input_tuple_desc : _output_tuple_desc; _total_rf_num = _parent->runtime_filter_num(); _is_load = (_input_tuple_desc != nullptr); + LOG(INFO) << "yy debug _is_load: " << _is_load; } Status VScanner::prepare(RuntimeState* state, const VExprContextSPtrs& conjuncts) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java index c212cda23dfdc58..300e252754ab0a7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java @@ -180,11 +180,6 @@ private void updateRequiredSlots() throws UserException { } // Update required slots and column_idxs in scanRangeLocations. setColumnPositionMapping(); - for (TScanRangeLocations location : scanRangeLocations) { - TFileScanRangeParams rangeParams = location.getScanRange().getExtScanRange().getFileScanRange().getParams(); - rangeParams.setRequiredSlots(params.getRequiredSlots()); - rangeParams.setColumnIdxs(params.getColumnIdxs()); - } } @Override @@ -430,3 +425,4 @@ protected static Optional getTFileType(String location) { } } + diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java index 0976cdb886fae4d..a81bd1984bee95c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java @@ -167,6 +167,9 @@ public class Coordinator { // copied from TQueryExecRequest; constant across all fragments private final TDescriptorTable descTable; + // scan node id -> TFileScanRangeParams + private Map fileScanRangeParamsMap = Maps.newHashMap(); + // Why do we use query global? // When `NOW()` function is in sql, we need only one now(), // but, we execute `NOW()` distributed. @@ -1940,7 +1943,7 @@ private void computeScanRangeAssignment() throws Exception { scanNodeIds.add(scanNode.getId().asInt()); if (scanNode instanceof FileQueryScanNode) { - fragmentExecParamsMap.get(scanNode.getFragmentId()).fileScanRangeParamsMap.put( + fileScanRangeParamsMap.put( scanNode.getId().asInt(), ((FileQueryScanNode) scanNode).getFileScanRangeParams()); } @@ -3071,8 +3074,6 @@ protected class FragmentExecParams { public List instanceExecParams = Lists.newArrayList(); public FragmentScanRangeAssignment scanRangeAssignment = new FragmentScanRangeAssignment(); - public Map fileScanRangeParamsMap = Maps.newHashMap(); - public FragmentExecParams(PlanFragment fragment) { this.fragment = fragment; } @@ -3435,3 +3436,4 @@ public FRuntimeFilterTargetParam(TUniqueId id, TNetworkAddress host) { } + diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java index 8a28c82ec95d2ce..7e1a4b66982c45c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java @@ -442,7 +442,6 @@ private void fillColumns(InternalService.PFetchTableSchemaResult result) private PFetchTableSchemaRequest getFetchTableStructureRequest() throws AnalysisException, TException { // set TFileScanRangeParams TFileScanRangeParams fileScanRangeParams = new TFileScanRangeParams(); - fileScanRangeParams.setFileType(getTFileType()); fileScanRangeParams.setFormatType(fileFormatType); fileScanRangeParams.setProperties(locationProperties); fileScanRangeParams.setFileAttributes(getFileAttributes()); @@ -466,9 +465,10 @@ private PFetchTableSchemaRequest getFetchTableStructureRequest() throws Analysis throw new AnalysisException("Can not get first file, please check uri."); } - fileScanRangeParams.setCompressType(Util.getOrInferCompressType(compressionType, firstFile.getPath())); // set TFileRangeDesc TFileRangeDesc fileRangeDesc = new TFileRangeDesc(); + fileRangeDesc.setFileType(getTFileType()); + fileRangeDesc.setCompressType(Util.getOrInferCompressType(compressionType, firstFile.getPath())); fileRangeDesc.setPath(firstFile.getPath()); fileRangeDesc.setStartOffset(0); fileRangeDesc.setSize(firstFile.getSize()); @@ -483,3 +483,4 @@ private PFetchTableSchemaRequest getFetchTableStructureRequest() throws Analysis } } + diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index a97fc364a1b95e4..cb0487bbdc7d6b0 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -405,7 +405,10 @@ struct TFileRangeDesc { // list: file location and range struct TFileScanRange { 1: optional list ranges - // deprecated, move to TDescriptorTbl + // If file_scan_params in TExecPlanFragmentParams is set in TExecPlanFragmentParams + // will use that field, otherwise, use this field. + // file_scan_params in TExecPlanFragmentParams will always be set in query request, + // and TFileScanRangeParams here is used for some other request such as fetch table schema for tvf. 2: optional TFileScanRangeParams params }