diff --git a/.github/workflows/slow_test.yml b/.github/workflows/slow_test.yml
index 6439887a4b..de45fa258d 100644
--- a/.github/workflows/slow_test.yml
+++ b/.github/workflows/slow_test.yml
@@ -126,7 +126,7 @@ jobs:
# Fix sanitizer: https://github.com/ClickHouse/ClickHouse/issues/64086
old_value=$(sudo sysctl -n vm.mmap_rnd_bits)
sudo sysctl -w vm.mmap_rnd_bits=28
- sudo docker exec ${TESTER_CONTAINER} bash -c "cd /infinity/ && rm -fr /var/infinity && export INF_DIRECTORY=`cat .tester_env` && echo INF_DIRECTORY=${INF_DIRECTORY} && python3 tools/run_cluster_test.py --infinity_path=cmake-build-debug/src/infinity --infinity_dir=${INF_DIRECTORY}"
+ sudo docker exec ${TESTER_CONTAINER} bash -c "cd /infinity/ && rm -fr /var/infinity && export INF_DIRECTORY=`cat .tester_env` && echo INF_DIRECTORY=${INF_DIRECTORY} && python3 tools/run_cluster_test.py --infinity_path=cmake-build-debug/src/infinity --infinity_dir=${INF_DIRECTORY} --minio_port=9005 --minio_console_port=9006"
sudo sysctl -w vm.mmap_rnd_bits=$old_value
- name: Collect thread sanitizer output in cluster test
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 5f839f347a..ebdade27b6 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -84,7 +84,7 @@ jobs:
id: run_cluster_test
run: |
sudo docker exec ${TESTER_CONTAINER} bash -c "rm -rf /root/.config/pip/pip.conf && cd /infinity/ && pip3 uninstall -y infinity-sdk infinity-embedded-sdk && cd python/infinity_sdk/ && pip3 install . -i https://pypi.tuna.tsinghua.edu.cn/simple --trusted-host tuna.tsinghua.edu.cn && cd ../.."
- sudo docker exec ${TESTER_CONTAINER} bash -c "cd /infinity/ && rm -fr /var/infinity && export INF_DIRECTORY=`cat .tester_env` && echo INF_DIRECTORY=${INF_DIRECTORY} && python3 tools/run_cluster_test.py --infinity_path=cmake-build-debug/src/infinity --infinity_dir=${INF_DIRECTORY}"
+ sudo docker exec ${TESTER_CONTAINER} bash -c "cd /infinity/ && rm -fr /var/infinity && export INF_DIRECTORY=`cat .tester_env` && echo INF_DIRECTORY=${INF_DIRECTORY} && python3 tools/run_cluster_test.py --infinity_path=cmake-build-debug/src/infinity --infinity_dir=${INF_DIRECTORY} --minio_port=9005 --minio_console_port=9006"
- name: Collect cluster test output
if: ${{ !cancelled() }}
@@ -123,7 +123,7 @@ jobs:
MINIO_DIR=~/minio_data_$(od -An -N4 -tx4 /dev/urandom | tr -d ' ')
echo "MINIO_CONTAINER=${MINIO_CONTAINER}" >> $GITHUB_ENV
echo "MINIO_DIR=${MINIO_DIR}" >> $GITHUB_ENV
- sudo docker rm -f -v ${MINIO_CONTAINER} && sudo rm -fr ${MINIO_DIR} && sudo mkdir ${MINIO_DIR} && sudo docker run -d --net=container:${BUILDER_CONTAINER} --name ${MINIO_CONTAINER} -e "MINIO_ROOT_PASSWORD=minioadmin" -e "MINIO_ROOT_USER=minioadmin" -v ${MINIO_DIR}:/data quay.io/minio/minio server /data --console-address ":9001" && sleep 5s
+ sudo docker rm -f -v ${MINIO_CONTAINER} && sudo rm -fr ${MINIO_DIR} && sudo mkdir ${MINIO_DIR} && sudo docker run -d --net=container:${BUILDER_CONTAINER} --name ${MINIO_CONTAINER} -e "MINIO_ROOT_PASSWORD=minioadmin" -e "MINIO_ROOT_USER=minioadmin" -v ${MINIO_DIR}:/data quay.io/minio/minio server /data --console-address ":9006" --address ":9005" && sleep 5s
- name: Start infinity debug version with minio
if: ${{ !cancelled() && !failure() }}
@@ -290,7 +290,7 @@ jobs:
MINIO_DIR=~/minio_data_$(od -An -N4 -tx4 /dev/urandom | tr -d ' ')
echo "MINIO_CONTAINER=${MINIO_CONTAINER}" >> $GITHUB_ENV
echo "MINIO_DIR=${MINIO_DIR}" >> $GITHUB_ENV
- sudo docker rm -f -v ${MINIO_CONTAINER} && sudo rm -fr ${MINIO_DIR} && sudo mkdir ${MINIO_DIR} && sudo docker run -d --net=container:${BUILDER_CONTAINER} --name ${MINIO_CONTAINER} -e "MINIO_ROOT_PASSWORD=minioadmin" -e "MINIO_ROOT_USER=minioadmin" -v ${MINIO_DIR}:/data quay.io/minio/minio server /data --console-address ":9001" && sleep 5s
+ sudo docker rm -f -v ${MINIO_CONTAINER} && sudo rm -fr ${MINIO_DIR} && sudo mkdir ${MINIO_DIR} && sudo docker run -d --net=container:${BUILDER_CONTAINER} --name ${MINIO_CONTAINER} -e "MINIO_ROOT_PASSWORD=minioadmin" -e "MINIO_ROOT_USER=minioadmin" -v ${MINIO_DIR}:/data quay.io/minio/minio server /data --console-address ":9006" --address ":9005" && sleep 5s
- name: Start infinity release version with minio
if: ${{ !cancelled() && !failure() }}
diff --git a/README.md b/README.md
index 9c85b75a2e..8ad93c72ea 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,5 @@
-
+
-
+
new_cell = MakeUnique| (lexeme);
if (size_ == 0) {
- head_ = new_cell;
- tail_ = new_cell;
+ Cell *cell_ptr = new_cell.release();
+ head_ = cell_ptr;
+ tail_ = cell_ptr;
size_++;
return true;
} else {
- if (tail_->CompareTo(new_cell) == 0) {
- delete new_cell;
+ if (tail_->CompareTo(new_cell.get()) == 0) {
return false;
- } else if (tail_->CompareTo(new_cell) < 0) {
- tail_->next_ = new_cell;
- new_cell->prev_ = tail_;
- tail_ = new_cell;
+ } else if (tail_->CompareTo(new_cell.get()) < 0) {
+ Cell *cell_ptr = new_cell.release();
+ tail_->next_ = cell_ptr;
+ cell_ptr->prev_ = tail_;
+ tail_ = cell_ptr;
size_++;
return true;
- } else if (head_->CompareTo(new_cell) > 0) {
- head_->prev_ = new_cell;
- new_cell->next_ = head_;
- head_ = new_cell;
+ } else if (head_->CompareTo(new_cell.get()) > 0) {
+ Cell *cell_ptr = new_cell.release();
+ head_->prev_ = cell_ptr;
+ cell_ptr->next_ = head_;
+ head_ = cell_ptr;
size_++;
return true;
} else {
Cell *index = tail_;
- while (index != nullptr && index->CompareTo(new_cell) > 0) {
+ while (index != nullptr && index->CompareTo(new_cell.get()) > 0) {
index = index->prev_;
}
- if (index->CompareTo(new_cell) == 0) {
- delete new_cell;
+ if (index->CompareTo(new_cell.get()) == 0) {
return false;
- } else if (index->CompareTo(new_cell) < 0) {
- new_cell->prev_ = index;
- new_cell->next_ = index->next_;
- index->next_->prev_ = new_cell;
- index->next_ = new_cell;
+ } else if (index->CompareTo(new_cell.get()) < 0) {
+ Cell *cell_ptr = new_cell.release();
+ cell_ptr->prev_ = index;
+ cell_ptr->next_ = index->next_;
+ index->next_->prev_ = cell_ptr;
+ index->next_ = cell_ptr;
size_++;
return true;
}
}
}
- delete new_cell;
return false;
}
} // namespace infinity
diff --git a/src/common/analyzer/keyword_analyzer.cpp b/src/common/analyzer/whitespace_analyzer.cpp
similarity index 89%
rename from src/common/analyzer/keyword_analyzer.cpp
rename to src/common/analyzer/whitespace_analyzer.cpp
index 9e539c44cc..fc76172a30 100644
--- a/src/common/analyzer/keyword_analyzer.cpp
+++ b/src/common/analyzer/whitespace_analyzer.cpp
@@ -16,15 +16,14 @@ module;
#include
#include
-module keyword_analyzer;
-
+module whitespace_analyzer;
import stl;
import term;
import analyzer;
namespace infinity {
-int KeywordAnalyzer::AnalyzeImpl(const Term &input, void *data, HookType func) {
+int WhitespaceAnalyzer::AnalyzeImpl(const Term &input, void *data, HookType func) {
std::istringstream is(input.text_);
std::string t;
u32 offset = 0;
diff --git a/src/common/analyzer/keyword_analyzer.cppm b/src/common/analyzer/whitespace_analyzer.cppm
similarity index 76%
rename from src/common/analyzer/keyword_analyzer.cppm
rename to src/common/analyzer/whitespace_analyzer.cppm
index bddf389714..d262b07c7f 100644
--- a/src/common/analyzer/keyword_analyzer.cppm
+++ b/src/common/analyzer/whitespace_analyzer.cppm
@@ -1,4 +1,4 @@
-// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
+// Copyright(C) 2024 InfiniFlow, Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -14,16 +14,17 @@
module;
-export module keyword_analyzer;
+export module whitespace_analyzer;
import stl;
import term;
import analyzer;
namespace infinity {
-export class KeywordAnalyzer : public Analyzer {
+
+export class WhitespaceAnalyzer : public Analyzer {
public:
- KeywordAnalyzer() = default;
- ~KeywordAnalyzer() override = default;
+ WhitespaceAnalyzer() = default;
+ ~WhitespaceAnalyzer() override = default;
protected:
int AnalyzeImpl(const Term &input, void *data, HookType func) override;
diff --git a/src/common/default_values.cppm b/src/common/default_values.cppm
index bbe4b4ac0d..d962ab620f 100644
--- a/src/common/default_values.cppm
+++ b/src/common/default_values.cppm
@@ -278,6 +278,9 @@ export {
constexpr std::string_view MEMINDEX_MEMORY_QUOTA_OPTION_NAME = "memindex_memory_quota";
constexpr std::string_view RESULT_CACHE_OPTION_NAME = "result_cache";
constexpr std::string_view CACHE_RESULT_CAPACITY_OPTION_NAME = "cache_result_capacity";
+ constexpr std::string_view DENSE_INDEX_BUILDING_WORKER_OPTION_NAME = "dense_index_building_worker";
+ constexpr std::string_view SPARSE_INDEX_BUILDING_WORKER_OPTION_NAME = "sparse_index_building_worker";
+ constexpr std::string_view FULLTEXT_INDEX_BUILDING_WORKER_OPTION_NAME = "fulltext_index_building_worker";
constexpr std::string_view WAL_DIR_OPTION_NAME = "wal_dir";
constexpr std::string_view WAL_COMPACT_THRESHOLD_OPTION_NAME = "wal_compact_threshold";
@@ -314,8 +317,10 @@ export {
constexpr std::string_view SYSTEM_MEMORY_USAGE_VAR_NAME = "system_memory_usage"; // global
constexpr std::string_view OPEN_FILE_COUNT_VAR_NAME = "open_file_count"; // global
constexpr std::string_view CPU_USAGE_VAR_NAME = "cpu_usage"; // global
- constexpr std::string_view FOLLOWER_NUMBER = "follower_number"; // global
+ constexpr std::string_view FOLLOWER_NUMBER_VAR_NAME = "follower_number"; // global
constexpr std::string_view CACHE_RESULT_NUM_VAR_NAME = "cache_result_num"; // global
+ constexpr std::string_view MEMORY_CACHE_MISS_VAR_NAME = "memory_cache_miss"; // global
+ constexpr std::string_view DISK_CACHE_MISS_VAR_NAME = "disk_cache_miss"; // global
// IO related
constexpr SizeT DEFAULT_READ_BUFFER_SIZE = 4096;
diff --git a/src/common/status.cpp b/src/common/status.cpp
index 158d1e7735..e5bfb74d7a 100644
--- a/src/common/status.cpp
+++ b/src/common/status.cpp
@@ -448,6 +448,10 @@ Status Status::FileIsOpen(const String &filename) { return Status(ErrorCode::kFi
Status Status::Unknown(const String &name) { return Status(ErrorCode::kUnknown, MakeUnique(fmt::format("Unknown {}", name))); }
+Status Status::InvalidQueryOption(const String &detail) {
+ return Status(ErrorCode::kUnknown, MakeUnique(fmt::format("Invalid query option: {}", detail)));
+}
+
// 4. TXN fail
Status Status::TxnRollback(u64 txn_id, const String &rollback_reason) {
return Status(ErrorCode::kTxnRollback, MakeUnique(fmt::format("Transaction: {} is rollback. {}", txn_id, rollback_reason)));
@@ -599,6 +603,12 @@ Status Status::NotRegistered(const String &node_info) {
Status Status::CantSwitchRole(const String &detailed_info) { return Status(ErrorCode::kCantSwitchRole, MakeUnique(detailed_info)); }
+Status Status::TooManyFollower(infinity::u8 follower_limit) {
+ return Status(ErrorCode::kTooManyFollower, MakeUnique(fmt::format("Too many followers, limit: {}", follower_limit)));
+}
+
+Status Status::TooManyLearner() { return Status(ErrorCode::kTooManyLearner, MakeUnique("Too many learner, limit: 255")); }
+
// meta
Status Status::InvalidEntry() { return Status(ErrorCode::kInvalidEntry, MakeUnique("Invalid entry")); }
diff --git a/src/common/status.cppm b/src/common/status.cppm
index 19bb71b379..b953e7d013 100644
--- a/src/common/status.cppm
+++ b/src/common/status.cppm
@@ -138,6 +138,7 @@ export enum class ErrorCode : long {
kErrorInit = 3089,
kFileIsOpen = 3090,
kUnknown = 3091,
+ kInvalidQueryOption = 3092,
// 4. Txn fail
kTxnRollback = 4001,
@@ -183,6 +184,8 @@ export enum class ErrorCode : long {
kInvalidStorageType = 7024,
kNotRegistered = 7025,
kCantSwitchRole = 7026,
+ kTooManyFollower = 7027,
+ kTooManyLearner = 7028,
// 8. meta error
kInvalidEntry = 8001,
@@ -311,6 +314,7 @@ public:
static Status ErrorInit(const String &detailed_info);
static Status FileIsOpen(const String &filename);
static Status Unknown(const String &name);
+ static Status InvalidQueryOption(const String& detail);
// 4. TXN fail
static Status TxnRollback(u64 txn_id, const String &rollback_reason = "no reanson gived");
@@ -356,6 +360,8 @@ public:
static Status InvalidStorageType(const String &expected, const String &actual);
static Status NotRegistered(const String &node_info);
static Status CantSwitchRole(const String &detailed_info);
+ static Status TooManyFollower(u8 follower_limit);
+ static Status TooManyLearner();
// meta
static Status InvalidEntry();
diff --git a/src/embedded_infinity/wrap_infinity.cpp b/src/embedded_infinity/wrap_infinity.cpp
index 93e1d4afca..a782925a16 100644
--- a/src/embedded_infinity/wrap_infinity.cpp
+++ b/src/embedded_infinity/wrap_infinity.cpp
@@ -1358,6 +1358,13 @@ void ProcessDataBlocks(QueryResult &query_result, WrapQueryResult &wrap_query_re
auto data_block = query_result.result_table_->GetDataBlockById(block_idx);
ProcessColumns(data_block, query_result.result_table_->ColumnCount(), columns);
}
+
+ if(query_result.result_table_->total_hits_count_flag_) {
+ nlohmann::json json_response;
+ json_response["total_hits_count"] = query_result.result_table_->total_hits_count_;
+ wrap_query_result.extra_result = json_response.dump();
+ }
+
HandleColumnDef(wrap_query_result, query_result.result_table_->ColumnCount(), query_result.result_table_->definition_ptr_, columns);
}
@@ -1368,6 +1375,7 @@ WrapQueryResult WrapSearch(Infinity &instance,
Vector highlight_list,
Vector order_by_list,
Vector group_by_list,
+ bool total_hits_count_flag,
WrapSearchExpr *wrap_search_expr,
WrapParsedExpr *filter_expr,
WrapParsedExpr *limit_expr,
@@ -1531,8 +1539,17 @@ WrapQueryResult WrapSearch(Infinity &instance,
}
}
- auto query_result =
- instance.Search(db_name, table_name, search_expr, filter, limit, offset, output_columns, highlight, order_by_exprs, group_by_exprs);
+ auto query_result = instance.Search(db_name,
+ table_name,
+ search_expr,
+ filter,
+ limit,
+ offset,
+ output_columns,
+ highlight,
+ order_by_exprs,
+ group_by_exprs,
+ total_hits_count_flag);
search_expr = nullptr;
filter = nullptr;
limit = nullptr;
diff --git a/src/embedded_infinity/wrap_infinity.cppm b/src/embedded_infinity/wrap_infinity.cppm
index 93e1147746..0e1181ef76 100644
--- a/src/embedded_infinity/wrap_infinity.cppm
+++ b/src/embedded_infinity/wrap_infinity.cppm
@@ -125,6 +125,7 @@ export struct WrapQueryResult {
Vector names;
Vector column_defs;
Vector column_fields;
+ String extra_result;
// show database
String database_name;
String store_dir;
@@ -432,6 +433,7 @@ export WrapQueryResult WrapSearch(Infinity &instance,
Vector highlight_list,
Vector order_by_list,
Vector group_by_list,
+ bool total_hits_count_flag,
WrapSearchExpr *wrap_search_expr = nullptr,
WrapParsedExpr *where_expr = nullptr,
WrapParsedExpr *limit_expr = nullptr,
diff --git a/src/embedded_infinity_ext.cpp b/src/embedded_infinity_ext.cpp
index cb92ed61b2..6124856de2 100644
--- a/src/embedded_infinity_ext.cpp
+++ b/src/embedded_infinity_ext.cpp
@@ -48,6 +48,7 @@ NB_MODULE(embedded_infinity_ext, m) {
.def_rw("names", &WrapQueryResult::names)
.def_rw("column_defs", &WrapQueryResult::column_defs)
.def_rw("column_fields", &WrapQueryResult::column_fields)
+ .def_rw("extra_result", &WrapQueryResult::extra_result)
.def_rw("database_name", &WrapQueryResult::database_name)
.def_rw("store_dir", &WrapQueryResult::store_dir)
.def_rw("table_count", &WrapQueryResult::table_count)
@@ -325,6 +326,7 @@ NB_MODULE(embedded_infinity_ext, m) {
nb::arg("highlight_list"),
nb::arg("order_by_list"),
nb::arg("group_by_list"),
+ nb::arg("total_hits_count_flag"),
nb::arg("wrap_search_expr") = nullptr,
nb::arg("where_expr") = nullptr,
nb::arg("limit_expr") = nullptr,
diff --git a/src/executor/operator/physical_compact_finish.cpp b/src/executor/operator/physical_compact_finish.cpp
index 852dbae8fa..341352b57e 100644
--- a/src/executor/operator/physical_compact_finish.cpp
+++ b/src/executor/operator/physical_compact_finish.cpp
@@ -33,6 +33,8 @@ import internal_types;
import infinity_context;
import infinity_exception;
import status;
+import txn_store;
+import segment_index_entry;
namespace infinity {
@@ -79,6 +81,29 @@ void PhysicalCompactFinish::SaveSegmentData(QueryContext *query_context, const C
}
LOG_DEBUG(ss.str());
+ for (const auto &compact_segment_data : compact_state_data->segment_data_list_) {
+ TxnStore *txn_store = txn->txn_store();
+ TxnTableStore *txn_table_store = txn_store->GetTxnTableStore(table_entry);
+ auto index_map = table_entry->IndexMetaMap();
+ for (const auto &[index_name, index_meta] : *index_map) {
+ auto [table_index_entry, status] = index_meta->GetEntryNolock(txn->TxnID(), txn->BeginTS());
+ if (!status.ok()) {
+ continue;
+ }
+ Vector segment_index_entries;
+ auto &segment_index_map = table_index_entry->index_by_segment();
+ for (const auto *old_segment : compact_segment_data.old_segments_) {
+ auto iter = segment_index_map.find(old_segment->segment_id());
+ if (iter == segment_index_map.end()) {
+ continue;
+ }
+ auto *segment_index_entry = iter->second.get();
+ segment_index_entries.push_back(segment_index_entry);
+ }
+ txn_table_store->AddSegmentIndexesStore(table_index_entry, std::move(segment_index_entries));
+ }
+ }
+
txn->Compact(table_entry, std::move(segment_data), compact_type_);
}
diff --git a/src/executor/operator/physical_import.cpp b/src/executor/operator/physical_import.cpp
index 1912b5ea76..62af0ee72b 100644
--- a/src/executor/operator/physical_import.cpp
+++ b/src/executor/operator/physical_import.cpp
@@ -1280,6 +1280,34 @@ void PhysicalImport::JSONLRowHandler(const nlohmann::json &line_json, Vector schema;
+ arrow::Status status = arrow_reader->GetSchema(&schema);
+ if (!status.ok()) {
+ return Status::ImportFileFormatError(status.ToString());
+ }
+ const arrow::FieldVector &fields = schema->fields();
+ const Vector> &column_defs = table_entry->column_defs();
+ if (fields.size() != column_defs.size()) {
+ return Status::ColumnCountMismatch(fmt::format("Column count mismatch: {} != {}", fields.size(), column_defs.size()));
+ }
+ for (SizeT i = 0; i < fields.size(); ++i) {
+ const auto &field = fields[i];
+ const auto &column_def = column_defs[i];
+
+ if (*column_def->type() != *field->type()) {
+ return Status::ImportFileFormatError(
+ fmt::format("Column {} mismatch, {} != {}", i, column_def->type()->ToString(), field->type()->ToString()));
+ }
+ }
+
+ return Status::OK();
+}
+
+} // namespace
+
void PhysicalImport::ImportPARQUET(QueryContext *query_context, ImportOperatorState *import_op_state) {
arrow::MemoryPool *pool = arrow::DefaultMemoryPool();
@@ -1304,6 +1332,10 @@ void PhysicalImport::ImportPARQUET(QueryContext *query_context, ImportOperatorSt
}
std::unique_ptr arrow_reader = build_result.MoveValueUnsafe();
+ if (Status status = CheckParquetColumns(table_entry_, arrow_reader.get()); !status.ok()) {
+ RecoverableError(status);
+ }
+
std::shared_ptr rb_reader;
if (auto status = arrow_reader->GetRecordBatchReader(&rb_reader); !status.ok()) {
RecoverableError(Status::ImportFileFormatError(status.ToString()));
@@ -1330,10 +1362,6 @@ void PhysicalImport::ImportPARQUET(QueryContext *query_context, ImportOperatorSt
auto batch = maybe_batch.MoveValueUnsafe();
const auto batch_row_count = batch->num_rows();
const auto batch_col_count = batch->num_columns();
- if (static_cast(batch_col_count) != table_entry_->ColumnCount()) {
- RecoverableError(
- Status::ColumnCountMismatch(fmt::format("Column count mismatch: {} != {}", batch_col_count, table_entry_->ColumnCount())));
- }
for (i64 batch_row_id = 0; batch_row_id < batch_row_count; ++batch_row_id) {
for (int column_idx = 0; column_idx < batch_col_count; ++column_idx) {
SharedPtr column = batch->column(column_idx);
@@ -1408,14 +1436,8 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
ColumnVector &column_vector,
i64 start_offset,
i64 end_offset) {
- if (sparse_info->DataType() != EmbeddingDataType::kElemBit && data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
switch (sparse_info->DataType()) {
case EmbeddingDataType::kElemBit: {
- if (data_array.get() != nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
ParquetSparseValueHandler(sparse_info,
index_array,
nullptr,
@@ -1425,13 +1447,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemUInt8: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto uint8_value_array = std::dynamic_pointer_cast(data_array->values());
- if (uint8_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto uint8_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
uint8_value_array,
@@ -1441,13 +1457,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemInt8: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto int8_value_array = std::dynamic_pointer_cast(data_array->values());
- if (int8_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int8_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
int8_value_array,
@@ -1457,13 +1467,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemInt16: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto int16_value_array = std::dynamic_pointer_cast(data_array->values());
- if (int16_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int16_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
int16_value_array,
@@ -1473,13 +1477,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemInt32: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto int32_value_array = std::dynamic_pointer_cast(data_array->values());
- if (int32_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int32_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
int32_value_array,
@@ -1489,13 +1487,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemInt64: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto int64_value_array = std::dynamic_pointer_cast(data_array->values());
- if (int64_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int64_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
int64_value_array,
@@ -1505,13 +1497,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemFloat16: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto float16_value_array = std::dynamic_pointer_cast(data_array->values());
- if (float16_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto float16_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
float16_value_array,
@@ -1521,13 +1507,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemBFloat16: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto float_value_array = std::dynamic_pointer_cast(data_array->values());
- if (float_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto float_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
float_value_array,
@@ -1537,13 +1517,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemFloat: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto float_value_array = std::dynamic_pointer_cast(data_array->values());
- if (float_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto float_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
float_value_array,
@@ -1553,13 +1527,7 @@ void ParquetSparseValueHandler(const SparseInfo *sparse_info,
break;
}
case EmbeddingDataType::kElemDouble: {
- if (data_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
- auto double_value_array = std::dynamic_pointer_cast(data_array->values());
- if (double_value_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto double_value_array = std::static_pointer_cast(data_array->values());
ParquetSparseValueHandler(sparse_info,
index_array,
double_value_array,
@@ -1595,10 +1563,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
const SizeT byte_size = dim / 8;
auto embedding = MakeUnique(byte_size);
- auto bool_array = std::dynamic_pointer_cast(list_array->values());
- if (bool_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto bool_array = std::static_pointer_cast(list_array->values());
auto *raw_u8_ptr = reinterpret_cast(embedding.get());
for (i64 j = start_offset; j < end_offset; ++j) {
if (bool_array->Value(j)) {
@@ -1610,10 +1575,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemUInt8: {
auto embedding = MakeUnique(dim * sizeof(u8));
- auto uint8_array = std::dynamic_pointer_cast(list_array->values());
- if (uint8_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto uint8_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
const u8 value = uint8_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1622,10 +1584,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemInt8: {
auto embedding = MakeUnique(dim * sizeof(i8));
- auto int8_array = std::dynamic_pointer_cast(list_array->values());
- if (int8_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int8_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
i8 value = int8_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1634,10 +1593,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemInt16: {
auto embedding = MakeUnique(dim * sizeof(i16));
- auto int16_array = std::dynamic_pointer_cast(list_array->values());
- if (int16_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int16_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
i16 value = int16_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1646,10 +1602,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemInt32: {
auto embedding = MakeUnique(dim * sizeof(i32));
- auto int32_array = std::dynamic_pointer_cast(list_array->values());
- if (int32_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int32_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
i32 value = int32_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1658,10 +1611,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemInt64: {
auto embedding = MakeUnique(dim * sizeof(i64));
- auto int64_array = std::dynamic_pointer_cast(list_array->values());
- if (int64_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto int64_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
i64 value = int64_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1670,10 +1620,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemFloat16: {
auto embedding = MakeUnique(dim * sizeof(Float16T));
- auto float16_array = std::dynamic_pointer_cast(list_array->values());
- if (float16_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto float16_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
const u16 value = float16_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1682,10 +1629,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemBFloat16: {
auto embedding = MakeUnique(dim * sizeof(BFloat16T));
- auto float_array = std::dynamic_pointer_cast(list_array->values());
- if (float_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto float_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
const float value = float_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1694,10 +1638,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemFloat: {
auto embedding = MakeUnique(dim * sizeof(float));
- auto float_array = std::dynamic_pointer_cast(list_array->values());
- if (float_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto float_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
float value = float_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1706,10 +1647,7 @@ Pair, SizeT> ParquetEmbeddingHandler(SharedPtr list_
}
case EmbeddingDataType::kElemDouble: {
auto embedding = MakeUnique(dim * sizeof(double));
- auto double_array = std::dynamic_pointer_cast(list_array->values());
- if (double_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto double_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
double value = double_array->Value(j);
reinterpret_cast(embedding.get())[j - start_offset] = value;
@@ -1736,10 +1674,7 @@ ParquetTensorHandler(SharedPtr list_array, const EmbeddingInfo
embedding_vec.push_back(std::move(data));
}
} else {
- auto tensor_ele_array = std::dynamic_pointer_cast(list_array->values());
- if (tensor_ele_array.get() == nullptr) {
- RecoverableError(Status::ImportFileFormatError("Invalid parquet file format."));
- }
+ auto tensor_ele_array = std::static_pointer_cast(list_array->values());
for (i64 j = start_offset; j < end_offset; ++j) {
auto data = ParquetEmbeddingHandler(tensor_ele_array, embedding_info, j);
embedding_vec.push_back(std::move(data));
@@ -1753,78 +1688,78 @@ ParquetTensorHandler(SharedPtr list_array, const EmbeddingInfo
void PhysicalImport::ParquetValueHandler(const SharedPtr &array, ColumnVector &column_vector, u64 value_idx) {
switch (const auto column_data_logical_type = column_vector.data_type()->type(); column_data_logical_type) {
case LogicalType::kBoolean: {
- auto value = std::dynamic_pointer_cast(array)->Value(value_idx);
+ auto value = std::static_pointer_cast(array)->Value(value_idx);
column_vector.AppendByPtr(reinterpret_cast(&value));
break;
}
case LogicalType::kTinyInt: {
- auto value = std::dynamic_pointer_cast(array)->Value(value_idx);
+ auto value = std::static_pointer_cast(array)->Value(value_idx);
column_vector.AppendByPtr(reinterpret_cast(&value));
break;
}
case LogicalType::kSmallInt: {
- auto value = std::dynamic_pointer_cast(array)->Value(value_idx);
+ auto value = std::static_pointer_cast(array)->Value(value_idx);
column_vector.AppendByPtr(reinterpret_cast(&value));
break;
}
case LogicalType::kInteger: {
- auto value = std::dynamic_pointer_cast(array)->Value(value_idx);
+ auto value = std::static_pointer_cast(array)->Value(value_idx);
column_vector.AppendByPtr(reinterpret_cast | |