From fcb898e9a484755710e8f53c18c36a701e53563d Mon Sep 17 00:00:00 2001
From: shuiyisong <113876041+shuiyisong@users.noreply.github.com>
Date: Thu, 13 Mar 2025 16:30:29 +0800
Subject: [PATCH] chore: support `inverted` index in pipeline (#5700)

chore: rebase main
---
 src/api/src/v1/column_def.rs                             | 9 +++++++++
 src/pipeline/src/etl/transform/index.rs                  | 5 +++++
 .../src/etl/transform/transformer/greptime/coerce.rs     | 7 +++++--
 tests-integration/tests/http.rs                          | 4 ++--
 4 files changed, 21 insertions(+), 4 deletions(-)

diff --git a/src/api/src/v1/column_def.rs b/src/api/src/v1/column_def.rs
index dab7784172f..a119bbb3e17 100644
--- a/src/api/src/v1/column_def.rs
+++ b/src/api/src/v1/column_def.rs
@@ -132,6 +132,15 @@ pub fn options_from_skipping(skipping: &SkippingIndexOptions) -> Result<Option<C
     Ok((!options.options.is_empty()).then_some(options))
 }
 
+/// Tries to construct a `ColumnOptions` for inverted index.
+pub fn options_from_inverted() -> ColumnOptions {
+    let mut options = ColumnOptions::default();
+    options
+        .options
+        .insert(INVERTED_INDEX_GRPC_KEY.to_string(), "true".to_string());
+    options
+}
+
 /// Tries to construct a `FulltextAnalyzer` from the given analyzer.
 pub fn as_fulltext_option(analyzer: Analyzer) -> FulltextAnalyzer {
     match analyzer {
diff --git a/src/pipeline/src/etl/transform/index.rs b/src/pipeline/src/etl/transform/index.rs
index 9b442ec220f..0300c568385 100644
--- a/src/pipeline/src/etl/transform/index.rs
+++ b/src/pipeline/src/etl/transform/index.rs
@@ -19,14 +19,17 @@ const INDEX_TIMEINDEX: &str = "time";
 const INDEX_TAG: &str = "tag";
 const INDEX_FULLTEXT: &str = "fulltext";
 const INDEX_SKIPPING: &str = "skipping";
+const INDEX_INVERTED: &str = "inverted";
 
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 #[allow(clippy::enum_variant_names)]
 pub enum Index {
     Time,
+    // deprecated, use Inverted instead
     Tag,
     Fulltext,
     Skipping,
+    Inverted,
 }
 
 impl std::fmt::Display for Index {
@@ -36,6 +39,7 @@ impl std::fmt::Display for Index {
             Index::Tag => INDEX_TAG,
             Index::Fulltext => INDEX_FULLTEXT,
             Index::Skipping => INDEX_SKIPPING,
+            Index::Inverted => INDEX_INVERTED,
         };
 
         write!(f, "{}", index)
@@ -59,6 +63,7 @@ impl TryFrom<&str> for Index {
             INDEX_TAG => Ok(Index::Tag),
             INDEX_FULLTEXT => Ok(Index::Fulltext),
             INDEX_SKIPPING => Ok(Index::Skipping),
+            INDEX_INVERTED => Ok(Index::Inverted),
             _ => UnsupportedIndexTypeSnafu { value }.fail(),
         }
     }
diff --git a/src/pipeline/src/etl/transform/transformer/greptime/coerce.rs b/src/pipeline/src/etl/transform/transformer/greptime/coerce.rs
index 97cefe48ff4..df6ab2fff9c 100644
--- a/src/pipeline/src/etl/transform/transformer/greptime/coerce.rs
+++ b/src/pipeline/src/etl/transform/transformer/greptime/coerce.rs
@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use api::v1::column_data_type_extension::TypeExt;
-use api::v1::column_def::{options_from_fulltext, options_from_skipping};
+use api::v1::column_def::{options_from_fulltext, options_from_inverted, options_from_skipping};
 use api::v1::{ColumnDataTypeExtension, ColumnOptions, JsonTypeExtension};
 use datatypes::schema::{FulltextOptions, SkippingIndexOptions};
 use greptime_proto::v1::value::ValueData;
@@ -102,7 +102,9 @@ fn coerce_semantic_type(transform: &Transform) -> SemanticType {
     match transform.index {
         Some(Index::Tag) => SemanticType::Tag,
         Some(Index::Time) => SemanticType::Timestamp,
-        Some(Index::Fulltext) | Some(Index::Skipping) | None => SemanticType::Field,
+        Some(Index::Fulltext) | Some(Index::Skipping) | Some(Index::Inverted) | None => {
+            SemanticType::Field
+        }
     }
 }
 
@@ -116,6 +118,7 @@ fn coerce_options(transform: &Transform) -> Result<Option<ColumnOptions>> {
         Some(Index::Skipping) => {
             options_from_skipping(&SkippingIndexOptions::default()).context(ColumnOptionsSnafu)
         }
+        Some(Index::Inverted) => Ok(Some(options_from_inverted())),
         _ => Ok(None),
     }
 }
diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs
index 0ad561c6015..d561580cc93 100644
--- a/tests-integration/tests/http.rs
+++ b/tests-integration/tests/http.rs
@@ -1265,6 +1265,7 @@ transform:
       - id1
       - id2
     type: int32
+    index: inverted
   - fields:
       - logger
     type: string
@@ -1350,8 +1351,7 @@ transform:
     assert_eq!(res.status(), StatusCode::OK);
 
     // 3. check schema
-
-    let expected_schema =   "[[\"logs1\",\"CREATE TABLE IF NOT EXISTS \\\"logs1\\\" (\\n  \\\"id1\\\" INT NULL,\\n  \\\"id2\\\" INT NULL,\\n  \\\"logger\\\" STRING NULL,\\n  \\\"type\\\" STRING NULL SKIPPING INDEX WITH(granularity = '10240', type = 'BLOOM'),\\n  \\\"log\\\" STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', case_sensitive = 'false'),\\n  \\\"time\\\" TIMESTAMP(9) NOT NULL,\\n  TIME INDEX (\\\"time\\\"),\\n  PRIMARY KEY (\\\"type\\\", \\\"log\\\")\\n)\\n\\nENGINE=mito\\nWITH(\\n  append_mode = 'true'\\n)\"]]";
+    let expected_schema = "[[\"logs1\",\"CREATE TABLE IF NOT EXISTS \\\"logs1\\\" (\\n  \\\"id1\\\" INT NULL INVERTED INDEX,\\n  \\\"id2\\\" INT NULL INVERTED INDEX,\\n  \\\"logger\\\" STRING NULL,\\n  \\\"type\\\" STRING NULL SKIPPING INDEX WITH(granularity = '10240', type = 'BLOOM'),\\n  \\\"log\\\" STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', case_sensitive = 'false'),\\n  \\\"time\\\" TIMESTAMP(9) NOT NULL,\\n  TIME INDEX (\\\"time\\\"),\\n  PRIMARY KEY (\\\"type\\\", \\\"log\\\")\\n)\\n\\nENGINE=mito\\nWITH(\\n  append_mode = 'true'\\n)\"]]";
     validate_data(
         "pipeline_schema",
         &client,