Skip to content

Commit 4c78d4d

Browse files
lxy-9602lxy-9602
committed
update lumina to 0.3.1
1 parent 6650c2f commit 4c78d4d

5 files changed

Lines changed: 140 additions & 52 deletions

File tree

‎src/paimon/global_index/lumina/lumina_global_index.cpp‎

Lines changed: 77 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -112,8 +112,12 @@ Result<LuminaTagField> ParseTagField(const rapidjson::Value& obj, const std::str
112112
}
113113

114114
LuminaTagField::ValueType parsed_value_type;
115-
if (value_type == std::string(::lumina::core::kExtensionTagVTypeInt64)) {
115+
if (value_type == std::string(::lumina::core::kExtensionTagVTypeInt32)) {
116+
parsed_value_type = LuminaTagField::ValueType::INT32;
117+
} else if (value_type == std::string(::lumina::core::kExtensionTagVTypeInt64)) {
116118
parsed_value_type = LuminaTagField::ValueType::INT64;
119+
} else if (value_type == std::string(::lumina::core::kExtensionTagVTypeFloat)) {
120+
parsed_value_type = LuminaTagField::ValueType::FLOAT;
117121
} else if (value_type == std::string(::lumina::core::kExtensionTagVTypeDouble)) {
118122
parsed_value_type = LuminaTagField::ValueType::DOUBLE;
119123
} else if (value_type == std::string(::lumina::core::kExtensionTagVTypeString)) {
@@ -122,11 +126,6 @@ Result<LuminaTagField> ParseTagField(const rapidjson::Value& obj, const std::str
122126
return Status::Invalid(fmt::format("lumina tag_schema {} has unsupported value_type: {}",
123127
tag_label, value_type));
124128
}
125-
if (parsed_type == LuminaTagField::Type::RANGE &&
126-
parsed_value_type == LuminaTagField::ValueType::STRING) {
127-
return Status::Invalid(fmt::format(
128-
"lumina tag_schema {} range type does not support string value_type", tag_label));
129-
}
130129
return LuminaTagField{key_name, parsed_type, parsed_value_type};
131130
}
132131

@@ -137,18 +136,22 @@ Status ValidateTagArrowType(const LuminaTagField& tag_field,
137136
value_type = list_type->value_type();
138137
}
139138

140-
// Lumina currently downcasts range tag values to 32-bit precision internally.
141-
// Reject Arrow int64/double inputs to avoid silent precision loss.
142139
bool compatible = false;
143140
switch (tag_field.value_type) {
144-
case LuminaTagField::ValueType::INT64:
141+
case LuminaTagField::ValueType::INT32:
145142
compatible = value_type->id() == arrow::Type::INT8 ||
146143
value_type->id() == arrow::Type::INT16 ||
147144
value_type->id() == arrow::Type::INT32;
148145
break;
149-
case LuminaTagField::ValueType::DOUBLE:
146+
case LuminaTagField::ValueType::INT64:
147+
compatible = value_type->id() == arrow::Type::INT64;
148+
break;
149+
case LuminaTagField::ValueType::FLOAT:
150150
compatible = value_type->id() == arrow::Type::FLOAT;
151151
break;
152+
case LuminaTagField::ValueType::DOUBLE:
153+
compatible = value_type->id() == arrow::Type::DOUBLE;
154+
break;
152155
case LuminaTagField::ValueType::STRING:
153156
compatible = value_type->id() == arrow::Type::STRING;
154157
break;
@@ -175,7 +178,7 @@ Status AppendTagValue(const std::shared_ptr<arrow::Array>& array, int64_t index,
175178
return Status::OK();
176179
}
177180

178-
if constexpr (std::is_same_v<ValueType, int64_t>) {
181+
if constexpr (std::is_same_v<ValueType, int32_t>) {
179182
switch (array->type_id()) {
180183
case arrow::Type::INT8:
181184
AppendPrimitiveTagValue<ValueType, arrow::Int8Array>(array, index, values);
@@ -191,16 +194,25 @@ Status AppendTagValue(const std::shared_ptr<arrow::Array>& array, int64_t index,
191194
fmt::format("lumina integer tag field has unsupported arrow type {}",
192195
array->type()->ToString()));
193196
}
197+
} else if constexpr (std::is_same_v<ValueType, int64_t>) {
198+
if (array->type_id() != arrow::Type::INT64) {
199+
return Status::Invalid(fmt::format(
200+
"lumina int64 tag field has unsupported arrow type {}", array->type()->ToString()));
201+
}
202+
AppendPrimitiveTagValue<ValueType, arrow::Int64Array>(array, index, values);
203+
} else if constexpr (std::is_same_v<ValueType, float>) {
204+
if (array->type_id() != arrow::Type::FLOAT) {
205+
return Status::Invalid(fmt::format(
206+
"lumina float tag field has unsupported arrow type {}", array->type()->ToString()));
207+
}
208+
AppendPrimitiveTagValue<ValueType, arrow::FloatArray>(array, index, values);
194209
} else if constexpr (std::is_same_v<ValueType, double>) {
195-
switch (array->type_id()) {
196-
case arrow::Type::FLOAT:
197-
AppendPrimitiveTagValue<ValueType, arrow::FloatArray>(array, index, values);
198-
break;
199-
default:
200-
return Status::Invalid(
201-
fmt::format("lumina floating tag field has unsupported arrow type {}",
202-
array->type()->ToString()));
210+
if (array->type_id() != arrow::Type::DOUBLE) {
211+
return Status::Invalid(
212+
fmt::format("lumina double tag field has unsupported arrow type {}",
213+
array->type()->ToString()));
203214
}
215+
AppendPrimitiveTagValue<ValueType, arrow::DoubleArray>(array, index, values);
204216
} else if constexpr (std::is_same_v<ValueType, std::string>) {
205217
if (array->type_id() != arrow::Type::STRING) {
206218
return Status::Invalid(
@@ -251,13 +263,17 @@ Result<TagValue> LiteralToTagValue(const Literal& literal) {
251263
}
252264
switch (literal.GetType()) {
253265
case FieldType::TINYINT:
254-
return TagValue(static_cast<int64_t>(literal.GetValue<int8_t>()));
266+
return TagValue(static_cast<int32_t>(literal.GetValue<int8_t>()));
255267
case FieldType::SMALLINT:
256-
return TagValue(static_cast<int64_t>(literal.GetValue<int16_t>()));
268+
return TagValue(static_cast<int32_t>(literal.GetValue<int16_t>()));
257269
case FieldType::INT:
258-
return TagValue(static_cast<int64_t>(literal.GetValue<int32_t>()));
270+
return TagValue(literal.GetValue<int32_t>());
271+
case FieldType::BIGINT:
272+
return TagValue(literal.GetValue<int64_t>());
259273
case FieldType::FLOAT:
260-
return TagValue(static_cast<double>(literal.GetValue<float>()));
274+
return TagValue(literal.GetValue<float>());
275+
case FieldType::DOUBLE:
276+
return TagValue(literal.GetValue<double>());
261277
case FieldType::STRING:
262278
return TagValue(literal.GetValue<std::string>());
263279
default:
@@ -285,6 +301,18 @@ Result<TagValues> LiteralsToTagValues(const std::vector<Literal>& literals) {
285301
case FieldType::TINYINT:
286302
case FieldType::SMALLINT:
287303
case FieldType::INT: {
304+
std::vector<int32_t> values;
305+
values.reserve(literals.size());
306+
for (const auto& literal : literals) {
307+
PAIMON_ASSIGN_OR_RAISE(TagValue value, LiteralToTagValue(literal));
308+
auto typed_value = std::get_if<int32_t>(&value);
309+
CHECK_NOT_NULL(typed_value,
310+
"lumina tag predicate IN literals must have the same value type");
311+
values.push_back(*typed_value);
312+
}
313+
return TagValues(std::move(values));
314+
}
315+
case FieldType::BIGINT: {
288316
std::vector<int64_t> values;
289317
values.reserve(literals.size());
290318
for (const auto& literal : literals) {
@@ -297,6 +325,18 @@ Result<TagValues> LiteralsToTagValues(const std::vector<Literal>& literals) {
297325
return TagValues(std::move(values));
298326
}
299327
case FieldType::FLOAT: {
328+
std::vector<float> values;
329+
values.reserve(literals.size());
330+
for (const auto& literal : literals) {
331+
PAIMON_ASSIGN_OR_RAISE(TagValue value, LiteralToTagValue(literal));
332+
auto typed_value = std::get_if<float>(&value);
333+
CHECK_NOT_NULL(typed_value,
334+
"lumina tag predicate IN literals must have the same value type");
335+
values.push_back(*typed_value);
336+
}
337+
return TagValues(std::move(values));
338+
}
339+
case FieldType::DOUBLE: {
300340
std::vector<double> values;
301341
values.reserve(literals.size());
302342
for (const auto& literal : literals) {
@@ -342,13 +382,27 @@ Result<std::vector<TagDimensionData>> LuminaIndexWriter::ExtractTagDataForSegmen
342382
TagDimensionData tag_dimension_data;
343383
tag_dimension_data.tagkName = tag_field.name;
344384
switch (tag_field.value_type) {
385+
case LuminaTagField::ValueType::INT32: {
386+
std::vector<std::vector<int32_t>> values;
387+
PAIMON_RETURN_NOT_OK(
388+
ExtractTagValues<int32_t>(field_array, segment_start, segment_len, &values));
389+
tag_dimension_data.values = std::move(values);
390+
break;
391+
}
345392
case LuminaTagField::ValueType::INT64: {
346393
std::vector<std::vector<int64_t>> values;
347394
PAIMON_RETURN_NOT_OK(
348395
ExtractTagValues<int64_t>(field_array, segment_start, segment_len, &values));
349396
tag_dimension_data.values = std::move(values);
350397
break;
351398
}
399+
case LuminaTagField::ValueType::FLOAT: {
400+
std::vector<std::vector<float>> values;
401+
PAIMON_RETURN_NOT_OK(
402+
ExtractTagValues<float>(field_array, segment_start, segment_len, &values));
403+
tag_dimension_data.values = std::move(values);
404+
break;
405+
}
352406
case LuminaTagField::ValueType::DOUBLE: {
353407
std::vector<std::vector<double>> values;
354408
PAIMON_RETURN_NOT_OK(

‎src/paimon/global_index/lumina/lumina_global_index.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,9 @@ struct LuminaTagField {
4444
};
4545

4646
enum class ValueType {
47+
INT32,
4748
INT64,
49+
FLOAT,
4850
DOUBLE,
4951
STRING,
5052
};

‎src/paimon/global_index/lumina/lumina_global_index_test.cpp‎

Lines changed: 54 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -306,7 +306,7 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithMixedTagPredicates) {
306306
std::map<std::string, std::string> tag_options = options_;
307307
tag_options["lumina.extension.build.tag.tag_schema"] =
308308
R"([{"key_name":"color","type":"enum","value_type":"string"},)"
309-
R"({"key_name":"price","type":"range","value_type":"double"}])";
309+
R"({"key_name":"price","type":"range","value_type":"float"}])";
310310

311311
std::shared_ptr<arrow::DataType> tag_data_type = arrow::struct_(
312312
{arrow::field("f0", arrow::list(arrow::float32())), arrow::field("color", arrow::utf8()),
@@ -354,7 +354,7 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithIntegerListTagFilter) {
354354

355355
std::map<std::string, std::string> tag_options = options_;
356356
tag_options["lumina.extension.build.tag.tag_schema"] =
357-
R"([{"key_name":"category_ids","type":"enum","value_type":"int64"}])";
357+
R"([{"key_name":"category_ids","type":"enum","value_type":"int32"}])";
358358

359359
std::shared_ptr<arrow::DataType> tag_data_type =
360360
arrow::struct_({arrow::field("f0", arrow::list(arrow::float32())),
@@ -392,22 +392,25 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithCompatibleTagArrowTypes) {
392392

393393
std::map<std::string, std::string> tag_options = options_;
394394
tag_options["lumina.extension.build.tag.tag_schema"] =
395-
R"([{"key_name":"tag_i8","type":"enum","value_type":"int64"},)"
396-
R"({"key_name":"tag_i16","type":"enum","value_type":"int64"},)"
397-
R"({"key_name":"tag_i32","type":"enum","value_type":"int64"},)"
398-
R"({"key_name":"tag_f32","type":"range","value_type":"double"}])";
395+
R"([{"key_name":"tag_i8","type":"enum","value_type":"int32"},)"
396+
R"({"key_name":"tag_i16","type":"enum","value_type":"int32"},)"
397+
R"({"key_name":"tag_i32","type":"enum","value_type":"int32"},)"
398+
R"({"key_name":"tag_i64","type":"enum","value_type":"int64"},)"
399+
R"({"key_name":"tag_f32","type":"range","value_type":"float"},)"
400+
R"({"key_name":"tag_f64","type":"enum","value_type":"double"}])";
399401

400402
std::shared_ptr<arrow::DataType> tag_data_type = arrow::struct_(
401403
{arrow::field("f0", arrow::list(arrow::float32())), arrow::field("tag_i8", arrow::int8()),
402404
arrow::field("tag_i16", arrow::int16()), arrow::field("tag_i32", arrow::int32()),
403-
arrow::field("tag_f32", arrow::float32())});
405+
arrow::field("tag_i64", arrow::int64()), arrow::field("tag_f32", arrow::float32()),
406+
arrow::field("tag_f64", arrow::float64())});
404407
std::shared_ptr<arrow::Array> tag_array =
405408
arrow::ipc::internal::json::ArrayFromJSON(tag_data_type,
406409
R"([
407-
[[0.0, 0.0, 0.0, 0.0], 1, 10, 100, 1.5],
408-
[[0.0, 1.0, 0.0, 1.0], 2, 20, 200, 2.5],
409-
[[1.0, 0.0, 1.0, 0.0], 3, 30, 300, 3.5],
410-
[[1.0, 1.0, 1.0, 1.0], 4, 40, 400, 4.5]
410+
[[0.0, 0.0, 0.0, 0.0], 1, 10, 100, 10000000001, 1.5, 1.25],
411+
[[0.0, 1.0, 0.0, 1.0], 2, 20, 200, 10000000002, 2.5, 2.25],
412+
[[1.0, 0.0, 1.0, 0.0], 3, 30, 300, 10000000003, 3.5, 3.25],
413+
[[1.0, 1.0, 1.0, 1.0], 4, 40, 400, 10000000004, 4.5, 4.25]
411414
])")
412415
.ValueOrDie();
413416

@@ -438,9 +441,24 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithCompatibleTagArrowTypes) {
438441
search_and_check(PredicateBuilder::Equal(/*field_index=*/3, /*field_name=*/"tag_i32",
439442
FieldType::INT, Literal(400)),
440443
{3l}, {0.01f});
441-
search_and_check(PredicateBuilder::GreaterThan(/*field_index=*/4, /*field_name=*/"tag_f32",
444+
search_and_check(PredicateBuilder::Equal(
445+
/*field_index=*/4, /*field_name=*/"tag_i64", FieldType::BIGINT,
446+
Literal(static_cast<int64_t>(10000000003))),
447+
{2l}, {2.21f});
448+
search_and_check(PredicateBuilder::In(
449+
/*field_index=*/4, /*field_name=*/"tag_i64", FieldType::BIGINT,
450+
{Literal(static_cast<int64_t>(10000000001)),
451+
Literal(static_cast<int64_t>(10000000004))}),
452+
{3l, 0l}, {0.01f, 4.21f});
453+
search_and_check(PredicateBuilder::GreaterThan(/*field_index=*/5, /*field_name=*/"tag_f32",
442454
FieldType::FLOAT, Literal(4.0f)),
443455
{3l}, {0.01f});
456+
search_and_check(PredicateBuilder::Equal(/*field_index=*/6, /*field_name=*/"tag_f64",
457+
FieldType::DOUBLE, Literal(3.25)),
458+
{2l}, {2.21f});
459+
search_and_check(PredicateBuilder::In(/*field_index=*/6, /*field_name=*/"tag_f64",
460+
FieldType::DOUBLE, {Literal(2.25), Literal(4.25)}),
461+
{3l, 1l}, {0.01f, 2.01f});
444462
}
445463

446464
TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithTagNullAndEmptyValues) {
@@ -452,10 +470,10 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithTagNullAndEmptyValues) {
452470
tag_options["lumina.extension.build.tag.tag_schema"] =
453471
R"([{"key_name":"color","type":"enum","value_type":"string"},)"
454472
R"({"key_name":"labels","type":"enum","value_type":"string"},)"
455-
R"({"key_name":"price","type":"range","value_type":"double"},)"
456-
R"({"key_name":"scores","type":"enum","value_type":"double"},)"
457-
R"({"key_name":"category","type":"enum","value_type":"int64"},)"
458-
R"({"key_name":"category_ids","type":"enum","value_type":"int64"}])";
473+
R"({"key_name":"price","type":"range","value_type":"float"},)"
474+
R"({"key_name":"scores","type":"enum","value_type":"float"},)"
475+
R"({"key_name":"category","type":"enum","value_type":"int32"},)"
476+
R"({"key_name":"category_ids","type":"enum","value_type":"int32"}])";
459477

460478
std::shared_ptr<arrow::DataType> tag_data_type = arrow::struct_(
461479
{arrow::field("f0", arrow::list(arrow::float32())), arrow::field("color", arrow::utf8()),
@@ -599,12 +617,13 @@ TEST_F(LuminaGlobalIndexTest, TestTagSchemaValidation) {
599617
R"([{"key_name":"color","type":"range","value_type":"string"}])";
600618
ASSERT_NOK_WITH_MSG(
601619
WriteGlobalIndex(index_root, tag_data_type, tag_options, array_, Range(0, 3)),
602-
"lumina tag_schema tag[0] range type does not support string value_type");
620+
"Option extension.build.tag.tag_schema tag[0] range type does not support value_type "
621+
"'string'");
603622
}
604623
{
605624
std::map<std::string, std::string> tag_options = options_;
606625
tag_options["lumina.extension.build.tag.tag_schema"] =
607-
R"([{"key_name":"color","type":"enum","value_type":"int64"}])";
626+
R"([{"key_name":"color","type":"enum","value_type":"int32"}])";
608627
ASSERT_NOK_WITH_MSG(
609628
WriteGlobalIndex(index_root, tag_data_type, tag_options, array_, Range(0, 3)),
610629
"lumina tag field color type string is not compatible with tag_schema value_type");
@@ -615,7 +634,7 @@ TEST_F(LuminaGlobalIndexTest, TestTagSchemaValidation) {
615634
arrow::field("category", arrow::int64())});
616635
std::map<std::string, std::string> tag_options = options_;
617636
tag_options["lumina.extension.build.tag.tag_schema"] =
618-
R"([{"key_name":"category","type":"enum","value_type":"int64"}])";
637+
R"([{"key_name":"category","type":"enum","value_type":"int32"}])";
619638
ASSERT_NOK_WITH_MSG(
620639
WriteGlobalIndex(index_root, int64_tag_data_type, tag_options, array_, Range(0, 3)),
621640
"lumina tag field category type int64 is not compatible with tag_schema value_type");
@@ -629,7 +648,20 @@ TEST_F(LuminaGlobalIndexTest, TestTagSchemaValidation) {
629648
R"([{"key_name":"price","type":"range","value_type":"double"}])";
630649
ASSERT_NOK_WITH_MSG(
631650
WriteGlobalIndex(index_root, double_tag_data_type, tag_options, array_, Range(0, 3)),
632-
"lumina tag field price type double is not compatible with tag_schema value_type");
651+
"Option extension.build.tag.tag_schema tag[0] range type does not support value_type "
652+
"'double'");
653+
}
654+
{
655+
std::shared_ptr<arrow::DataType> int64_tag_data_type =
656+
arrow::struct_({arrow::field("f0", arrow::list(arrow::float32())),
657+
arrow::field("price", arrow::int64())});
658+
std::map<std::string, std::string> tag_options = options_;
659+
tag_options["lumina.extension.build.tag.tag_schema"] =
660+
R"([{"key_name":"price","type":"range","value_type":"int64"}])";
661+
ASSERT_NOK_WITH_MSG(
662+
WriteGlobalIndex(index_root, int64_tag_data_type, tag_options, array_, Range(0, 3)),
663+
"Option extension.build.tag.tag_schema tag[0] range type does not support value_type "
664+
"'int64'");
633665
}
634666
}
635667

@@ -644,8 +676,8 @@ TEST_F(LuminaGlobalIndexTest, TestGetExtraFieldNames) {
644676
std::map<std::string, std::string> tag_options = options_;
645677
tag_options["lumina.extension.build.tag.tag_schema"] =
646678
R"([{"key_name":"color","type":"enum","value_type":"string"},)"
647-
R"({"key_name":"price","type":"range","value_type":"double"},)"
648-
R"({"key_name":"category_ids","type":"enum","value_type":"int64"}])";
679+
R"({"key_name":"price","type":"range","value_type":"float"},)"
680+
R"({"key_name":"category_ids","type":"enum","value_type":"int32"}])";
649681
LuminaGlobalIndex global_index(tag_options);
650682
ASSERT_OK_AND_ASSIGN(std::optional<std::vector<std::string>> field_names,
651683
global_index.GetExtraFieldNames());

0 commit comments

Comments
 (0)