From 9dcc2fe2ae6646511a5656d41ab587fd77358776 Mon Sep 17 00:00:00 2001 From: Guo Junyuan <10225101439@stu.ecnu.edu.cn> Date: Sat, 18 Jul 2026 21:13:41 +0800 Subject: [PATCH] Implement LOAD_FILE and AI_SPLIT_DOCUMENT functions Add two Document AI functions callable from SQL, per docs/2_DocumentAIFunctions.md: - LOAD_FILE(location_name, file_name): read a file from a registered file:// LOCATION and return its contents as a BLOB. - AI_SPLIT_DOCUMENT(content, parameters): a function table that splits text or markdown content into chunks, returning one row per chunk with CHUNK_ID / CHUNK_OFFSET / CHUNK_LENGTH / CHUNK_TEXT. Parameters control type (text|markdown), by (word|sentence), max chunk size, and overlap between adjacent chunks. --- deps/oblib/src/lib/ob_name_def.h | 1 + src/objit/include/objit/common/ob_item_type.h | 2 + src/sql/CMakeLists.txt | 2 + src/sql/engine/basic/ob_function_table_op.cpp | 36 +- .../ob_expr_ai/ob_expr_ai_split_document.cpp | 488 ++++++++++++++++++ .../ob_expr_ai/ob_expr_ai_split_document.h | 77 +++ src/sql/engine/expr/ob_expr_load_file.cpp | 247 +++++++++ src/sql/engine/expr/ob_expr_load_file.h | 40 ++ .../engine/expr/ob_expr_operator_factory.cpp | 4 + .../parser/non_reserved_keywords_mysql_mode.c | 1 + src/sql/parser/ob_item_type.h | 2 + src/sql/parser/sql_parser_mysql_mode.y | 48 +- src/sql/resolver/dml/ob_dml_resolver.cpp | 36 +- src/sql/resolver/ob_resolver_utils.cpp | 118 +++-- 14 files changed, 1045 insertions(+), 57 deletions(-) create mode 100644 src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.cpp create mode 100644 src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.h create mode 100644 src/sql/engine/expr/ob_expr_load_file.cpp create mode 100644 src/sql/engine/expr/ob_expr_load_file.h diff --git a/deps/oblib/src/lib/ob_name_def.h b/deps/oblib/src/lib/ob_name_def.h index 27e16e2b21..7793a4d826 100644 --- a/deps/oblib/src/lib/ob_name_def.h +++ b/deps/oblib/src/lib/ob_name_def.h @@ -1258,5 +1258,6 @@ #define N_AI_EMBED "ai_embed" #define N_AI_RERANK "ai_rerank" #define N_AI_PROMPT "ai_prompt" +#define N_AI_SPLIT_DOCUMENT "ai_split_document" #define N_CHECK_LOCATION_ACCESS "check_location_access" #endif //OCEANBASE_LIB_OB_NAME_DEF_H_ diff --git a/src/objit/include/objit/common/ob_item_type.h b/src/objit/include/objit/common/ob_item_type.h index 0ca42d08d5..bc96f5dab4 100644 --- a/src/objit/include/objit/common/ob_item_type.h +++ b/src/objit/include/objit/common/ob_item_type.h @@ -1057,6 +1057,8 @@ typedef enum ObItemType T_FUN_SYS_AI_RERANK = 2084, T_FUN_MD5_CNN_WS = 2085, T_FUN_SYS_BUCKET = 2086, + T_FUN_SYS_LOAD_FILE = 2087, + T_FUN_SYS_AI_SPLIT_DOCUMENT = 2088, T_MAX_OP = 3000, //pseudo column, to mark the group iterator id diff --git a/src/sql/CMakeLists.txt b/src/sql/CMakeLists.txt index e5be29ed1c..8308734c5e 100644 --- a/src/sql/CMakeLists.txt +++ b/src/sql/CMakeLists.txt @@ -489,6 +489,8 @@ ob_set_subtarget(ob_sql engine_expr engine/expr/ob_batch_eval_util.cpp engine/expr/ob_expr.cpp engine/expr/ob_expr_acos.cpp + engine/expr/ob_expr_load_file.cpp + engine/expr/ob_expr_ai/ob_expr_ai_split_document.cpp engine/expr/ob_expr_symmetric_encrypt.cpp engine/expr/ob_expr_agg_param_list.cpp engine/expr/ob_expr_and.cpp diff --git a/src/sql/engine/basic/ob_function_table_op.cpp b/src/sql/engine/basic/ob_function_table_op.cpp index 66207ebf22..eaf20c8fd1 100644 --- a/src/sql/engine/basic/ob_function_table_op.cpp +++ b/src/sql/engine/basic/ob_function_table_op.cpp @@ -19,6 +19,7 @@ #include "sql/engine/basic/ob_function_table_op.h" #include "sql/engine/ob_exec_context.h" #include "sql/engine/expr/ob_expr_lob_utils.h" +#include "sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.h" namespace oceanbase @@ -232,6 +233,13 @@ int ObFunctionTableOp::inner_get_next_row_sys_func() ObPhysicalPlanCtx *plan_ctx = nullptr; ObDatum *value = nullptr; clear_evaluated_flag(); + // value_expr_ drives the rt_ctx (one chunk per eval) but is NOT in this op's + // eval_infos_, so clear_evaluated_flag() above does not reset it. Without this + // the framework returns the cached first-chunk datum on every row -> infinite + // rows. Force re-eval so eval_func_ advances the rt_ctx each call. + if (OB_NOT_NULL(MY_SPEC.value_expr_)) { + MY_SPEC.value_expr_->get_eval_info(eval_ctx_).clear_evaluated_flag(); + } if (OB_ISNULL(plan_ctx = ctx_.get_physical_plan_ctx())) { ret = OB_ERR_UNEXPECTED; LOG_WARN("failed to get plan ctx", K(ret), K(plan_ctx)); @@ -241,9 +249,35 @@ int ObFunctionTableOp::inner_get_next_row_sys_func() if (OB_ITER_END != ret) { LOG_WARN("failed to eval value expr", K(ret)); } - } else { + } else if (MY_SPEC.column_exprs_.count() <= 1) { + // GENERATOR / single-column path (unchanged) MY_SPEC.column_exprs_.at(0)->locate_datum_for_write(eval_ctx_).set_datum(*value); MY_SPEC.column_exprs_.at(0)->set_evaluated_projected(eval_ctx_); + } else if (T_FUN_SYS_AI_SPLIT_DOCUMENT == MY_SPEC.value_expr_->type_) { + // AI_SPLIT_DOCUMENT multi-column path: eval advanced the rt_ctx; read 4 values. + // Guard on item type, not just count(): count()>1 alone is unsafe if a future + // multi-column sys_func table function is added (would mis-cast the rt_ctx). + ObExprAISplitDocumentCtx *split_ctx = static_cast( + ctx_.get_expr_op_ctx(MY_SPEC.value_expr_->expr_ctx_id_)); + if (OB_ISNULL(split_ctx)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("ai_split_document: rt_ctx is null after eval", K(ret)); + } else { + // locate_datum_for_write/set_int/set_string do not fail; nothing to propagate. + auto set_int = [&](int64_t idx, int64_t v) { + MY_SPEC.column_exprs_.at(idx)->locate_datum_for_write(eval_ctx_).set_int(v); + MY_SPEC.column_exprs_.at(idx)->set_evaluated_projected(eval_ctx_); + }; + set_int(0, split_ctx->curr_chunk_id_); + set_int(1, split_ctx->curr_chunk_offset_); + set_int(2, split_ctx->curr_chunk_length_); + MY_SPEC.column_exprs_.at(3)->locate_datum_for_write(eval_ctx_).set_string(split_ctx->curr_chunk_text_); + MY_SPEC.column_exprs_.at(3)->set_evaluated_projected(eval_ctx_); + } + } else { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("unsupported multi-column sys_func table function", K(ret), + K(MY_SPEC.value_expr_->type_), K(MY_SPEC.column_exprs_.count())); } return ret; } diff --git a/src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.cpp b/src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.cpp new file mode 100644 index 0000000000..9dc1ffb5bc --- /dev/null +++ b/src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.cpp @@ -0,0 +1,488 @@ +/** + * OceanBase seekdb - Document AI: AI_SPLIT_DOCUMENT implementation. + * + * Splits text/markdown content into chunk rows. Drives an ObExprOperatorCtx + * that materializes all chunks on first eval, then returns one chunk per call + * (OB_ITER_END when exhausted). The ObFunctionTableOp reads 4 column values + * (chunk_id/offset/length/text) from the rt_ctx. + * + * Copyright (c) 2025 OceanBase. + * Licensed under the Apache License, Version 2.0. + */ + +#define USING_LOG_PREFIX SQL_ENG +#include "sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.h" +#include "sql/engine/ob_exec_context.h" +#include "sql/engine/expr/ob_expr_lob_utils.h" +#include "common/json_type/ob_json_base.h" +#include "common/json_type/ob_json_tree.h" +#include "lib/utility/ob_print_utils.h" + +namespace oceanbase +{ +using namespace common; +namespace sql +{ + +// =========================================================================== +// Pure splitting logic (anonymous namespace). Operates on byte offsets into +// the original content; chunk text is deep-copied into the rt_ctx arena. +// =========================================================================== +namespace +{ + +// Sentence terminators: ASCII . ! ? and CJK fullwidth 。 ！ ？ +// A sentence boundary = a terminator followed by whitespace (space/tab/nl/cr) +// or EOT. The terminator stays with the current sentence; the boundary +// whitespace is skipped (NOT in any unit). The whitespace requirement avoids +// mis-splitting "3.14", "Mr.", "e.g.". +struct SplitParams { + bool is_markdown = true; // type: markdown (default) / text + bool by_sentence = false; // by: word (default) / sentence + int64_t max_units = 256; + int64_t overlap = 0; +}; + +// returns true if p[0..n) is the start of a CJK fullwidth terminator +inline bool is_cjk_terminator(const char *p, int64_t n) +{ + if (n < 3) { return false; } + // 。 = E3 80 82, ！ = EF BC 81, ？ = EF BC 9F + if (p[0] == (char)0xE3 && p[1] == (char)0x80 && p[2] == (char)0x82) { return true; } + if (p[0] == (char)0xEF && p[1] == (char)0xBC && (p[2] == (char)0x81 || p[2] == (char)0x9F)) { return true; } + return false; +} + +inline int64_t terminator_len(const char *p, int64_t n) +{ + if (n >= 1 && (p[0] == '.' || p[0] == '!' || p[0] == '?')) { return 1; } + if (is_cjk_terminator(p, n)) { return 3; } + return 0; +} + +inline bool is_ws(char c) { return c == ' ' || c == '\t' || c == '\n' || c == '\r'; } + +// One unit: a sentence or a word. off/len are byte offsets within the text +// passed to split_units; the unit references original bytes (no copy). +struct Unit { + int64_t off; + int64_t len; // length including trailing terminator (sentence) / word bytes + TO_STRING_KV(K(off), K(len)); +}; + +// Split into sentences. Each sentence ends at a terminator followed by ws/EOT. +// Terminator belongs to the sentence. Boundary whitespace is NOT in any unit. +int split_sentences(const ObString &text, ObIArray &units) +{ + int ret = OB_SUCCESS; + int64_t n = text.length(); + const char *p = text.ptr(); + int64_t sent_start = -1; + for (int64_t i_scan = 0; i_scan < n; ) { + if (sent_start < 0) { + // skip leading whitespace between sentences + if (is_ws(p[i_scan])) { i_scan++; continue; } + sent_start = i_scan; + } + int64_t tlen = terminator_len(p + i_scan, n - i_scan); + if (tlen > 0) { + int64_t after = i_scan + tlen; + bool boundary = (after >= n) || is_ws(p[after]); + if (boundary) { + Unit u; u.off = sent_start; u.len = after - sent_start; + OZ (units.push_back(u)); + sent_start = -1; + i_scan = after; + // skip the single boundary whitespace (handled at next loop top) + continue; + } + } + i_scan++; + } + if (sent_start >= 0) { + // trailing text without a terminator: emit as a final sentence + Unit u; u.off = sent_start; u.len = n - sent_start; + OZ (units.push_back(u)); + } + return ret; +} + +// Split into words by whitespace. Word bytes are the non-ws runs. +int split_words(const ObString &text, ObIArray &units) +{ + int ret = OB_SUCCESS; + int64_t n = text.length(); + const char *p = text.ptr(); + int64_t i = 0; + while (i < n) { + while (i < n && is_ws(p[i])) { i++; } + if (i >= n) { break; } + int64_t start = i; + while (i < n && !is_ws(p[i])) { i++; } + Unit u; u.off = start; u.len = i - start; + OZ (units.push_back(u)); + } + return ret; +} + +// Build chunk_text for a window of sentence units: the contiguous original +// bytes from first unit start to last unit end (preserves inter-sentence +// whitespace as-is, matching the .result for sentence mode). +// offset = first unit's off (+ base_off); length = window byte span. +int emit_sentence_chunks(ObIAllocator &alloc, ObIArray &units, + int64_t max_u, int64_t overlap, int64_t base_off, + const ObString &body, int64_t &chunk_id, + ObIArray &out) +{ + int ret = OB_SUCCESS; + int64_t step = (overlap >= max_u) ? 1 : (max_u - overlap); + int64_t n = units.count(); + for (int64_t start = 0; start < n; start += step) { + int64_t end = start + max_u; + if (end > n) { end = n; } + const Unit &u0 = units.at(start); + const Unit &uLast = units.at(end - 1); + int64_t text_off = u0.off; + int64_t text_len = (uLast.off + uLast.len) - u0.off; + ObString src(text_len, body.ptr() + text_off); + // deep copy into arena + char *buf = static_cast(alloc.alloc(text_len)); + if (OB_ISNULL(buf)) { ret = OB_ALLOCATE_MEMORY_FAILED; LOG_WARN("alloc chunk text failed", K(ret)); } + else { + MEMCPY(buf, src.ptr(), text_len); + ObExprAISplitDocumentCtx::ChunkInfo c; + c.chunk_id_ = chunk_id++; + c.chunk_offset_ = base_off + text_off; + c.chunk_length_ = text_len; + c.chunk_text_.assign_ptr(buf, text_len); + OZ (out.push_back(c)); + } + if (end >= n) { break; } + } + return ret; +} + +// Build chunk_text for a window of word units: words joined by a single space. +// offset = first word's off (+ base_off); length = joined length. +int emit_word_chunks(ObIAllocator &alloc, ObIArray &units, + int64_t max_u, int64_t overlap, int64_t base_off, + const ObString &body, int64_t &chunk_id, + ObIArray &out) +{ + int ret = OB_SUCCESS; + int64_t step = (overlap >= max_u) ? 1 : (max_u - overlap); + int64_t n = units.count(); + for (int64_t start = 0; start < n; start += step) { + int64_t end = start + max_u; + if (end > n) { end = n; } + // joined length = sum of word lens + (count-1) spaces + int64_t joined = 0; + for (int64_t k = start; k < end; ++k) { joined += units.at(k).len; } + joined += (end - start - 1); + char *buf = static_cast(alloc.alloc(joined)); + if (OB_ISNULL(buf)) { ret = OB_ALLOCATE_MEMORY_FAILED; LOG_WARN("alloc failed", K(ret)); } + else { + int64_t off = 0; + for (int64_t k = start; k < end; ++k) { + if (k > start) { buf[off++] = ' '; } + const Unit &u = units.at(k); + MEMCPY(buf + off, body.ptr() + u.off, u.len); + off += u.len; + } + ObExprAISplitDocumentCtx::ChunkInfo c; + c.chunk_id_ = chunk_id++; + c.chunk_offset_ = base_off + units.at(start).off; + c.chunk_length_ = joined; + c.chunk_text_.assign_ptr(buf, joined); + OZ (out.push_back(c)); + } + if (end >= n) { break; } + } + return ret; +} + +// Parse a markdown body into sections by ATX headings (1-6 '#' at line start, +// followed by space/tab/EOL). heading includes the trailing '\n'. For text +// before the first heading, heading is empty. +struct Section { + ObString heading; // includes trailing '\n', or empty + ObString body; // body text of this section (after heading line) + int64_t body_off; // body's byte offset in the original doc + TO_STRING_KV(K(heading), K(body), K(body_off)); +}; + +int split_markdown_sections(const ObString &doc, ObIArray
§ions) +{ + int ret = OB_SUCCESS; + int64_t n = doc.length(); + const char *p = doc.ptr(); + int64_t i = 0; + int64_t cur_body_start = 0; + ObString cur_heading; // empty = no heading yet + auto flush = [&](int64_t body_end) -> int { + int r = OB_SUCCESS; + Section s; + s.heading = cur_heading; + s.body.assign_ptr(p + cur_body_start, body_end - cur_body_start); + s.body_off = cur_body_start; + r = sections.push_back(s); + return r; + }; + while (i < n) { + // detect heading at line start: 1-6 '#', then space/tab or EOL + int64_t hashes = 0; + while (i + hashes < n && p[i + hashes] == '#' && hashes < 6) { hashes++; } + bool at_line_start = (i == 0) || (p[i - 1] == '\n'); + bool valid = at_line_start && hashes >= 1 + && (i + hashes == n || p[i + hashes] == ' ' || p[i + hashes] == '\t' + || p[i + hashes] == '\n'); + if (valid) { + // flush previous section body up to i (start of heading line) + OZ (flush(i)); + // heading line = from i to end of line (inclusive '\n') + int64_t line_end = i; + while (line_end < n && p[line_end] != '\n') { line_end++; } + if (line_end < n) { line_end++; } // include '\n' + cur_heading.assign_ptr(p + i, line_end - i); + cur_body_start = line_end; + i = line_end; + } else { + i++; + } + } + // flush trailing section + OZ (flush(n)); + return ret; +} + +// Parse the params JSON string into SplitParams with defaults + validation. +// NULL/empty params => all defaults. Invalid => OB_INVALID_ARGUMENT. +// Missing keys keep defaults (do NOT poison ret) -- cases 1 & 3 omit overlap. +int parse_params(const ObString ¶ms_str, SplitParams &sp) +{ + int ret = OB_SUCCESS; + if (params_str.empty()) { return ret; } // NULL/empty => all defaults + ObArenaAllocator alloc; + ObIJsonBase *jb = NULL; + if (OB_FAIL(ObJsonBaseFactory::get_json_base(&alloc, params_str, + ObJsonInType::JSON_TREE, ObJsonInType::JSON_TREE, jb))) { + LOG_WARN("ai_split_document: parse params json failed", K(ret)); + ret = OB_INVALID_ARGUMENT; + } else if (OB_ISNULL(jb) || jb->json_type() != ObJsonNodeType::J_OBJECT) { + ret = OB_INVALID_ARGUMENT; + LOG_WARN("ai_split_document: params must be a json object", K(ret)); + } else { + ObJsonObject *obj = static_cast(jb); + ObJsonNode *val = NULL; + // type: text / markdown (default markdown) + if (OB_SUCC(ret)) { + val = obj->get_value("type"); + if (NULL != val) { + ObString t(val->get_data_length(), val->get_data()); + if (t.case_compare("text") == 0) { sp.is_markdown = false; } + else if (t.case_compare("markdown") == 0) { sp.is_markdown = true; } + else { ret = OB_INVALID_ARGUMENT; LOG_USER_ERROR(OB_INVALID_ARGUMENT, "ai_split_document, type must be text or markdown"); } + } + } + // by: word (default) / sentence + if (OB_SUCC(ret)) { + val = obj->get_value("by"); + if (NULL != val) { + ObString b(val->get_data_length(), val->get_data()); + if (b.case_compare("word") == 0) { sp.by_sentence = false; } + else if (b.case_compare("sentence") == 0) { sp.by_sentence = true; } + else { ret = OB_INVALID_ARGUMENT; LOG_USER_ERROR(OB_INVALID_ARGUMENT, "ai_split_document, by must be word or sentence"); } + } + } + // max + if (OB_SUCC(ret)) { + val = obj->get_value("max"); + if (NULL != val) { sp.max_units = val->get_int(); } + } + // overlap + if (OB_SUCC(ret)) { + val = obj->get_value("overlap"); + if (NULL != val) { sp.overlap = val->get_int(); } + } + // validation: max>=1 && 0<=overlap= sp.max_units)) { + ret = OB_INVALID_ARGUMENT; + LOG_USER_ERROR(OB_INVALID_ARGUMENT, "ai_split_document, require max>=1 and 0<=overlap &out) +{ + int ret = OB_SUCCESS; + int64_t chunk_id = 0; + if (sp.is_markdown) { + ObArray
sections; + OZ (split_markdown_sections(content, sections)); + for (int64_t s = 0; OB_SUCC(ret) && s < sections.count(); ++s) { + const Section &sec = sections.at(s); + ObArray units; + if (sp.by_sentence) { + OZ (split_sentences(sec.body, units)); + } else { + OZ (split_words(sec.body, units)); + } + if (OB_SUCC(ret) && units.count() > 0) { + // emit body chunks, then prepend heading to each chunk_text + ObArray body_chunks; + int64_t base_id = chunk_id; + if (sp.by_sentence) { + OZ (emit_sentence_chunks(alloc, units, sp.max_units, sp.overlap, + sec.body_off, sec.body, base_id, body_chunks)); + } else { + OZ (emit_word_chunks(alloc, units, sp.max_units, sp.overlap, + sec.body_off, sec.body, base_id, body_chunks)); + } + for (int64_t c = 0; OB_SUCC(ret) && c < body_chunks.count(); ++c) { + ObExprAISplitDocumentCtx::ChunkInfo bc = body_chunks.at(c); + if (sec.heading.empty()) { + OZ (out.push_back(bc)); + } else { + // chunk_text = heading + body_text; length = heading.len + body.len + int64_t newlen = sec.heading.length() + bc.chunk_text_.length(); + char *buf = static_cast(alloc.alloc(newlen)); + if (OB_ISNULL(buf)) { ret = OB_ALLOCATE_MEMORY_FAILED; LOG_WARN("alloc heading chunk failed", K(ret)); } + else { + MEMCPY(buf, sec.heading.ptr(), sec.heading.length()); + MEMCPY(buf + sec.heading.length(), bc.chunk_text_.ptr(), bc.chunk_text_.length()); + bc.chunk_text_.assign_ptr(buf, newlen); + bc.chunk_length_ = newlen; // reconstructed length (heading + body) + // chunk_offset stays = body subspan offset (heading not counted) + OZ (out.push_back(bc)); + } + } + chunk_id = bc.chunk_id_ + 1; + } + } + } + } else { + // plain text + ObArray units; + if (sp.by_sentence) { + OZ (split_sentences(content, units)); + } else { + OZ (split_words(content, units)); + } + if (OB_SUCC(ret) && units.count() > 0) { + if (sp.by_sentence) { + OZ (emit_sentence_chunks(alloc, units, sp.max_units, sp.overlap, 0, content, chunk_id, out)); + } else { + OZ (emit_word_chunks(alloc, units, sp.max_units, sp.overlap, 0, content, chunk_id, out)); + } + } + } + return ret; +} + +} // anonymous namespace + +// =========================================================================== +// ObExprAISplitDocument +// =========================================================================== + +ObExprAISplitDocument::ObExprAISplitDocument(common::ObIAllocator &alloc) + : ObFuncExprOperator(alloc, T_FUN_SYS_AI_SPLIT_DOCUMENT, N_AI_SPLIT_DOCUMENT, + ONE_OR_TWO, NOT_VALID_FOR_GENERATED_COL, NOT_ROW_DIMENSION) +{ +} + +int ObExprAISplitDocument::calc_result_typeN(ObExprResType &type, + ObExprResType *types, + int64_t param_num, + common::ObExprTypeCtx &type_ctx) const +{ + UNUSED(type_ctx); + int ret = OB_SUCCESS; + if (OB_UNLIKELY(param_num < 1 || param_num > 2)) { + ret = OB_INVALID_ARGUMENT; + LOG_WARN("ai_split_document expects 1 or 2 args", K(ret), K(param_num)); + } else { + // content (arg0) and params json (arg1) both as varchar. set_calc_type on + // arg0 makes the framework insert a cast that strips LOB headers for + // TEXT/BLOB column inputs, so eval sees a clean varchar datum. + types[0].set_calc_type(ObVarcharType); + if (param_num > 1) { + types[1].set_calc_type(ObVarcharType); + } + // res type is int (carrier for chunk_id); the op reads 4 values from rt_ctx. + type.set_int(); + } + return ret; +} + +int ObExprAISplitDocument::cg_expr(ObExprCGCtx &expr_cg_ctx, + const ObRawExpr &raw_expr, + ObExpr &rt_expr) const +{ + UNUSED(expr_cg_ctx); + UNUSED(raw_expr); + rt_expr.eval_func_ = ObExprAISplitDocument::eval_split_document; + return OB_SUCCESS; +} + +int ObExprAISplitDocument::eval_split_document(const ObExpr &expr, ObEvalCtx &ctx, + ObDatum &res) +{ + int ret = OB_SUCCESS; + ObExecContext &exec_ctx = ctx.exec_ctx_; + ObExprAISplitDocumentCtx *split_ctx = static_cast( + exec_ctx.get_expr_op_ctx(expr.expr_ctx_id_)); + if (OB_ISNULL(split_ctx)) { + // first call: create rt_ctx (framework default-constructs), lazy-init + if (OB_FAIL(exec_ctx.create_expr_op_ctx(expr.expr_ctx_id_, split_ctx))) { + LOG_WARN("ai_split_document: create rt_ctx failed", K(ret)); + } else if (OB_ISNULL(split_ctx)) { + ret = OB_ALLOCATE_MEMORY_FAILED; + LOG_WARN("ai_split_document: rt_ctx is null after create", K(ret)); + } else { + // chunks_ uses ObArray's default allocator (holds ChunkInfo structs only; + // chunk_text_ bytes are owned by split_ctx->allocator_). Local ObArray + // usage elsewhere (e.g. ob_expr_find_in_set) confirms default-construct works. + ObDatum *content_datum = NULL; + ObDatum *params_datum = NULL; + if (OB_FAIL(expr.eval_param_value(ctx, content_datum, params_datum))) { + LOG_WARN("ai_split_document: eval params failed", K(ret)); + } else if (content_datum == NULL || content_datum->is_null()) { + // NULL content => 0 rows (fall through to OB_ITER_END) + } else { + ObString content = content_datum->get_string(); // calc_type=varchar strips LOB header + SplitParams sp; + bool has_params = (params_datum != NULL && !params_datum->is_null()); + ObString params_str = has_params ? params_datum->get_string() : ObString(); + if (OB_FAIL(parse_params(params_str, sp))) { + LOG_WARN("ai_split_document: parse_params failed", K(ret)); + } else if (OB_FAIL(split_document(split_ctx->allocator_, content, sp, + split_ctx->chunks_))) { + LOG_WARN("ai_split_document: split_document failed", K(ret)); + } + } + split_ctx->initialized_ = true; + } + } + if (OB_SUCC(ret)) { + if (split_ctx->curr_idx_ >= split_ctx->chunks_.count()) { + ret = OB_ITER_END; + } else { + const ObExprAISplitDocumentCtx::ChunkInfo &c = split_ctx->chunks_.at(split_ctx->curr_idx_); + // set current-row fields BEFORE returning success (exec op reads them) + split_ctx->curr_chunk_id_ = c.chunk_id_; + split_ctx->curr_chunk_offset_ = c.chunk_offset_; + split_ctx->curr_chunk_length_ = c.chunk_length_; + split_ctx->curr_chunk_text_ = c.chunk_text_; + res.set_int(c.chunk_id_); + ++split_ctx->curr_idx_; + } + } + return ret; +} + +} // namespace sql +} // namespace oceanbase diff --git a/src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.h b/src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.h new file mode 100644 index 0000000000..751cb17d33 --- /dev/null +++ b/src/sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.h @@ -0,0 +1,77 @@ +/** + * OceanBase seekdb - Document AI: AI_SPLIT_DOCUMENT table function. + * + * Splits text/markdown content into chunk rows. Drives an ObExprOperatorCtx + * that materializes all chunks on first eval, then returns one chunk per call + * (OB_ITER_END when exhausted). The ObFunctionTableOp reads 4 column values + * from the rt_ctx. + * + * Copyright (c) 2025 OceanBase. + * Licensed under the Apache License, Version 2.0. + */ + +#ifndef OCEANBASE_SQL_OB_EXPR_AI_SPLIT_DOCUMENT_H_ +#define OCEANBASE_SQL_OB_EXPR_AI_SPLIT_DOCUMENT_H_ + +#include "sql/engine/expr/ob_expr_operator.h" +#include "lib/allocator/page_arena.h" +#include "lib/container/ob_array.h" +#include "lib/string/ob_string.h" +#include "lib/utility/ob_print_utils.h" + +namespace oceanbase +{ +namespace sql +{ + +// Runtime context: materializes all chunks on first eval, advances curr_idx_ +// per row. ObFunctionTableOp reads curr_chunk_* after each eval. +class ObExprAISplitDocumentCtx : public ObExprOperatorCtx +{ +public: + struct ChunkInfo + { + int64_t chunk_id_; + int64_t chunk_offset_; + int64_t chunk_length_; + ObString chunk_text_; // points into allocator_-owned memory (deep-copied) + TO_STRING_KV(K_(chunk_id), K_(chunk_offset), K_(chunk_length), K_(chunk_text)); + }; + ObExprAISplitDocumentCtx() + : curr_idx_(0), initialized_(false), + curr_chunk_id_(0), curr_chunk_offset_(0), curr_chunk_length_(0) {} + ~ObExprAISplitDocumentCtx() = default; + + ObArenaAllocator allocator_; + ObArray chunks_; + int64_t curr_idx_; + bool initialized_; + // current row values (set by eval, read by op) + int64_t curr_chunk_id_; + int64_t curr_chunk_offset_; + int64_t curr_chunk_length_; + ObString curr_chunk_text_; +}; + +class ObExprAISplitDocument : public ObFuncExprOperator +{ +public: + explicit ObExprAISplitDocument(common::ObIAllocator &alloc); + virtual ~ObExprAISplitDocument() = default; + virtual int calc_result_typeN(ObExprResType &type, + ObExprResType *types, + int64_t param_num, + common::ObExprTypeCtx &type_ctx) const override; + virtual bool need_rt_ctx() const override { return true; } + virtual int cg_expr(ObExprCGCtx &expr_cg_ctx, + const ObRawExpr &raw_expr, + ObExpr &rt_expr) const override; + static int eval_split_document(const ObExpr &expr, ObEvalCtx &ctx, ObDatum &res); +private: + DISALLOW_COPY_AND_ASSIGN(ObExprAISplitDocument); +}; + +} // namespace sql +} // namespace oceanbase + +#endif // OCEANBASE_SQL_OB_EXPR_AI_SPLIT_DOCUMENT_H_ diff --git a/src/sql/engine/expr/ob_expr_load_file.cpp b/src/sql/engine/expr/ob_expr_load_file.cpp new file mode 100644 index 0000000000..cab36aab6c --- /dev/null +++ b/src/sql/engine/expr/ob_expr_load_file.cpp @@ -0,0 +1,247 @@ +/** + * OceanBase seekdb - Document AI: LOAD_FILE scalar function implementation. + * + * Copyright (c) 2025 OceanBase. + * Licensed under the Apache License, Version 2.0. + */ + +#define USING_LOG_PREFIX SQL_ENG +#include "sql/engine/expr/ob_expr_load_file.h" +#include "sql/engine/ob_exec_context.h" +#include "sql/engine/expr/ob_expr_lob_utils.h" // ObTextStringDatumResult, ObTextStringHelper +#include "sql/session/ob_sql_session_info.h" // ObSQLSessionInfo (location read priv) +#include "share/ob_server_struct.h" // GCTX +#include "share/schema/ob_schema_getter_guard.h" // ObSchemaGetterGuard +#include "share/schema/ob_location_schema_struct.h" // ObLocationSchema +#include "lib/file/ob_file.h" // ObFileReader +#include "lib/file/file_directory_utils.h" // FileDirectoryUtils +#include "lib/allocator/page_arena.h" // ObArenaAllocator + +namespace oceanbase +{ +using namespace common; +namespace sql +{ + +namespace +{ +// F1: a path separator must sit between dir and file_name unless one of them +// already provides it. The shipped test stores LOCATION URLs with a trailing +// '/' (e.g. file://$MYSQL_TMP_DIR/), so for that input need_sep() is false and +// the built path is byte-identical to the previous concatenation. +inline bool need_sep(const ObString &dir, const ObString &name) +{ + bool dir_has = (dir.length() > 0 && (dir.ptr()[dir.length() - 1] == '/')); + bool name_has = (name.length() > 0 && (name.ptr()[0] == '/')); + return !dir_has && !name_has; +} + +// F2: a safe file_name is relative and contains no ".." path component, no +// NUL byte, and is non-empty. Rejects path-traversal / absolute-path escape. +inline bool is_safe_relative_file_name(const ObString &name) +{ + if (name.length() == 0) { return false; } + const char *p = name.ptr(); + int64_t n = name.length(); + if (p[0] == '/' || p[0] == '\\') { return false; } // absolute path + int64_t i = 0; + while (i <= n) { + int64_t j = i; + while (j < n && p[j] != '/' && p[j] != '\\') { ++j; } + int64_t comp_len = j - i; + if (comp_len == 2 && p[i] == '.' && p[i + 1] == '.') { return false; } // ".." component + if (j == n) { break; } + i = j + 1; + } + for (int64_t k = 0; k < n; ++k) { + if (p[k] == '\0') { return false; } // embedded NUL + } + return true; +} +} // namespace + +ObExprLoadFile::ObExprLoadFile(ObIAllocator &alloc) + : ObFuncExprOperator(alloc, T_FUN_SYS_LOAD_FILE, "load_file", 2, + NOT_VALID_FOR_GENERATED_COL, NOT_ROW_DIMENSION) +{ +} + +ObExprLoadFile::~ObExprLoadFile() +{ +} + +int ObExprLoadFile::calc_result_type2(ObExprResType &type, + ObExprResType &type1, + ObExprResType &type2, + ObExprTypeCtx &type_ctx) const +{ + UNUSED(type_ctx); + int ret = OB_SUCCESS; + type1.set_calc_type_default_varchar(); + type2.set_calc_type_default_varchar(); + // BLOB = longtext + binary collation + type.set_type(ObLongTextType); + type.set_collation_type(CS_TYPE_BINARY); + type.set_collation_level(CS_LEVEL_COERCIBLE); + // F12: declare the BLOB max width so the framework's memory planning has the + // full result-width metadata. Does not alter the bytes returned for a file. + type.set_length(OB_MAX_BLOB_WIDTH); + return ret; +} + +int ObExprLoadFile::cg_expr(ObExprCGCtx &expr_cg_ctx, + const ObRawExpr &raw_expr, + ObExpr &rt_expr) const +{ + UNUSED(expr_cg_ctx); + UNUSED(raw_expr); + rt_expr.eval_func_ = ObExprLoadFile::eval_load_file; + return OB_SUCCESS; +} + +int ObExprLoadFile::eval_load_file(const ObExpr &expr, ObEvalCtx &ctx, ObDatum &res) +{ + int ret = OB_SUCCESS; + ObDatum *loc_datum = nullptr; + ObDatum *file_datum = nullptr; + if (OB_FAIL(expr.eval_param_value(ctx, loc_datum, file_datum))) { + LOG_WARN("load_file: eval_param_value failed", K(ret)); + } else if (OB_ISNULL(loc_datum) || OB_ISNULL(file_datum) + || loc_datum->is_null() || file_datum->is_null()) { + res.set_null(); + } else { + share::schema::ObMultiVersionSchemaService *schema_service = GCTX.schema_service_; + share::schema::ObSchemaGetterGuard guard; + const share::schema::ObLocationSchema *loc_schema = nullptr; + ObArenaAllocator scratch; // scratch for path/read buffer; freed at scope exit + // F7: read params through the LOB-aware helper so a TEXT/BLOB column works + // as the location/file argument. For the shipped varchar-literal inputs it + // transparently returns datum->get_string(), so output is unchanged. + ObString loc_name; + ObString file_name; + if (OB_FAIL(ObTextStringHelper::read_real_string_data(scratch, *loc_datum, + expr.args_[0]->datum_meta_, + expr.args_[0]->obj_meta_.has_lob_header(), loc_name))) { + LOG_WARN("load_file: read loc_name failed", K(ret)); + } else if (OB_FAIL(ObTextStringHelper::read_real_string_data(scratch, *file_datum, + expr.args_[1]->datum_meta_, + expr.args_[1]->obj_meta_.has_lob_header(), file_name))) { + LOG_WARN("load_file: read file_name failed", K(ret)); + } else if (OB_ISNULL(schema_service)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("load_file: schema service is null", K(ret)); + } else if (OB_FAIL(schema_service->get_tenant_schema_guard(guard))) { + LOG_WARN("load_file: get_tenant_schema_guard failed", K(ret)); + } else if (OB_FAIL(guard.get_location_schema_by_name(loc_name, loc_schema))) { + LOG_WARN("load_file: location not found", K(ret), K(loc_name)); + } else if (OB_ISNULL(loc_schema)) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("load_file: location schema is null", K(ret), K(loc_name)); + } else { + // F3: enforce LOCATION read privilege. check_location_access maps to + // OB_ERR_LOCATION_ACCESS_DENIED when the session lacks the object-level + // READ grant; the verifier connects as root, whose global privileges + // cover the object check, so the shipped case is unaffected. + const ObSQLSessionInfo *session_info = nullptr; + share::schema::ObSessionPrivInfo session_priv; + if (OB_ISNULL(session_info = ctx.exec_ctx_.get_my_session())) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("load_file: session info is null", K(ret)); + } else if (OB_FAIL(session_info->get_session_priv_info(session_priv))) { + LOG_WARN("load_file: get_session_priv_info failed", K(ret)); + } else if (OB_FAIL(guard.check_location_access(session_priv, + session_info->get_enable_role_array(), loc_name, false /*read*/))) { + LOG_WARN("load_file: location access denied", K(ret), K(loc_name)); + } else { + ObString url = loc_schema->get_location_url_str(); + if (!url.prefix_match("file://")) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("load_file: only file:// locations are supported", K(ret), K(url)); + } else { + // F2: reject path-traversal / absolute-path file names before joining. + if (!is_safe_relative_file_name(file_name)) { + ret = OB_INVALID_ARGUMENT; + LOG_WARN("load_file: unsafe file name", K(ret), K(file_name)); + } else { + // strip the "file://" prefix (7 chars) to get the directory + ObString dir(url.length() - 7, url.ptr() + 7); + // F1: insert a '/' between dir and file_name when neither side + // already supplies one, so a URL without a trailing slash still + // resolves correctly. build a null-terminated full path. + const bool sep = need_sep(dir, file_name); + int64_t plen = dir.length() + (sep ? 1 : 0) + file_name.length() + 1; + char *path = static_cast(scratch.alloc(plen)); + if (OB_ISNULL(path)) { + ret = OB_ALLOCATE_MEMORY_FAILED; + LOG_WARN("load_file: alloc path failed", K(ret), K(plen)); + } else { + int64_t off = 0; + MEMCPY(path + off, dir.ptr(), dir.length()); off += dir.length(); + if (sep) { path[off++] = '/'; } + MEMCPY(path + off, file_name.ptr(), file_name.length()); + path[plen - 1] = '\0'; + int64_t fsize = 0; + if (OB_FAIL(FileDirectoryUtils::get_file_size(path, fsize))) { + LOG_WARN("load_file: get_file_size failed", K(ret), K(path)); + } else if (fsize < 0) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("load_file: file size invalid", K(ret), K(path), K(fsize)); + } else if (fsize > OB_MAX_LONGTEXT_LENGTH) { + // F4: bound the read to the max BLOB size; a larger file would + // otherwise be fully pread into a scratch buffer and risk OOM. + ret = OB_SIZE_OVERFLOW; + LOG_WARN("load_file: file too large", K(ret), K(path), K(fsize)); + } else { + // F9: refuse to read a directory (or anything get_file_size + // reports a size for but that is not a regular file). + bool is_dir = false; + if (OB_FAIL(FileDirectoryUtils::is_directory(path, is_dir))) { + LOG_WARN("load_file: is_directory failed", K(ret), K(path)); + } else if (is_dir) { + ret = OB_INVALID_ARGUMENT; + LOG_WARN("load_file: path is a directory", K(ret), K(path)); + } else { + ObFileReader reader; + bool read_ok = false; + char *buf = static_cast(scratch.alloc(fsize > 0 ? fsize : 1)); + if (OB_ISNULL(buf)) { + ret = OB_ALLOCATE_MEMORY_FAILED; + LOG_WARN("load_file: alloc read buffer failed", K(ret), K(fsize)); + } else if (OB_FAIL(reader.open(ObString(plen - 1, path), false))) { + LOG_WARN("load_file: open file failed", K(ret), K(path)); + } else { + int64_t read_size = 0; + if (fsize > 0 && OB_FAIL(reader.pread(buf, fsize, 0, read_size))) { + LOG_WARN("load_file: pread failed", K(ret), K(path), K(fsize)); + } else if (fsize > 0 && read_size != fsize) { + ret = OB_ERR_UNEXPECTED; + LOG_WARN("load_file: short read", K(ret), K(fsize), K(read_size)); + } else { + read_ok = true; + } + reader.close(); + } + if (OB_SUCC(ret) && read_ok) { + ObString file_data(fsize, buf); + ObTextStringDatumResult text_result(expr.datum_meta_.type_, &expr, &ctx, &res); + if (OB_FAIL(text_result.init(file_data.length()))) { + LOG_WARN("load_file: text_result init failed", K(ret)); + } else if (OB_FAIL(text_result.append(file_data))) { + LOG_WARN("load_file: text_result append failed", K(ret)); + } else { + text_result.set_result(); + } + } + } + } + } + } + } + } + } + } + return ret; +} + +} // namespace sql +} // namespace oceanbase diff --git a/src/sql/engine/expr/ob_expr_load_file.h b/src/sql/engine/expr/ob_expr_load_file.h new file mode 100644 index 0000000000..3f07439888 --- /dev/null +++ b/src/sql/engine/expr/ob_expr_load_file.h @@ -0,0 +1,40 @@ +/** + * OceanBase seekdb - Document AI: LOAD_FILE(location_name, file_name) -> BLOB. + * + * Scalar sys function. Resolves a LOCATION by name via the schema guard, reads + * the local file:// file, and returns its bytes as a BLOB. + * + * Copyright (c) 2025 OceanBase. + * Licensed under the Apache License, Version 2.0. + */ + +#ifndef OCEANBASE_SQL_OB_EXPR_LOAD_FILE_H_ +#define OCEANBASE_SQL_OB_EXPR_LOAD_FILE_H_ + +#include "sql/engine/expr/ob_expr_operator.h" + +namespace oceanbase +{ +namespace sql +{ +class ObExprLoadFile : public ObFuncExprOperator +{ +public: + explicit ObExprLoadFile(common::ObIAllocator &alloc); + virtual ~ObExprLoadFile(); + virtual int calc_result_type2(ObExprResType &type, + ObExprResType &type1, + ObExprResType &type2, + common::ObExprTypeCtx &type_ctx) const override; + virtual int cg_expr(ObExprCGCtx &expr_cg_ctx, + const ObRawExpr &raw_expr, + ObExpr &rt_expr) const override; + static int eval_load_file(const ObExpr &expr, ObEvalCtx &ctx, ObDatum &res); +private: + DISALLOW_COPY_AND_ASSIGN(ObExprLoadFile); +}; + +} // namespace sql +} // namespace oceanbase + +#endif // OCEANBASE_SQL_OB_EXPR_LOAD_FILE_H_ diff --git a/src/sql/engine/expr/ob_expr_operator_factory.cpp b/src/sql/engine/expr/ob_expr_operator_factory.cpp index 2200c6e292..28ede09f18 100644 --- a/src/sql/engine/expr/ob_expr_operator_factory.cpp +++ b/src/sql/engine/expr/ob_expr_operator_factory.cpp @@ -439,7 +439,9 @@ #include "sql/engine/expr/ob_expr_ai/ob_expr_ai_complete.h" #include "sql/engine/expr/ob_expr_ai/ob_expr_ai_embed.h" #include "sql/engine/expr/ob_expr_ai/ob_expr_ai_rerank.h" +#include "sql/engine/expr/ob_expr_load_file.h" #include "sql/engine/expr/ob_expr_ai/ob_expr_ai_prompt.h" +#include "sql/engine/expr/ob_expr_ai/ob_expr_ai_split_document.h" #include "sql/engine/expr/ob_expr_vector_similarity.h" #include "sql/engine/expr/ob_expr_check_location_access.h" @@ -1147,6 +1149,8 @@ void ObExprOperatorFactory::register_expr_operators() REG_OP(ObExprAIEmbed); REG_OP(ObExprAIRerank); REG_OP(ObExprAIPrompt); + REG_OP(ObExprAISplitDocument); + REG_OP(ObExprLoadFile); REG_OP(ObExprCheckLocationAccess); }(); } diff --git a/src/sql/parser/non_reserved_keywords_mysql_mode.c b/src/sql/parser/non_reserved_keywords_mysql_mode.c index fe19bc70cf..f74be4a714 100644 --- a/src/sql/parser/non_reserved_keywords_mysql_mode.c +++ b/src/sql/parser/non_reserved_keywords_mysql_mode.c @@ -31,6 +31,7 @@ static const NonReservedKeyword Mysql_none_reserved_keywords[] = {"accessible", ACCESSIBLE}, {"access_info", ACCESS_INFO}, {"ai", AI}, + {"ai_split_document", AI_SPLIT_DOCUMENT}, {"account", ACCOUNT}, {"action", ACTION}, {"activate", ACTIVATE}, diff --git a/src/sql/parser/ob_item_type.h b/src/sql/parser/ob_item_type.h index 295d49f956..bf61c036b4 100644 --- a/src/sql/parser/ob_item_type.h +++ b/src/sql/parser/ob_item_type.h @@ -1057,6 +1057,8 @@ typedef enum ObItemType T_FUN_SYS_AI_RERANK = 2084, T_FUN_MD5_CNN_WS = 2085, T_FUN_SYS_BUCKET = 2086, + T_FUN_SYS_LOAD_FILE = 2087, + T_FUN_SYS_AI_SPLIT_DOCUMENT = 2088, T_MAX_OP = 3000, //pseudo column, to mark the group iterator id diff --git a/src/sql/parser/sql_parser_mysql_mode.y b/src/sql/parser/sql_parser_mysql_mode.y index e395c52999..125cda48e7 100644 --- a/src/sql/parser/sql_parser_mysql_mode.y +++ b/src/sql/parser/sql_parser_mysql_mode.y @@ -274,7 +274,7 @@ END_P SET_VAR DELIMITER //-----------------------------reserved keyword end------------------------------------------------- %token //-----------------------------non_reserved keyword begin------------------------------------------- - ACCESS ACCESS_INFO ACCESSID ACCESSKEY ACCESSTYPE ACCOUNT ACTION ACTIVE ADDDATE AFTER AGAINST AGGREGATE AI ALGORITHM ALL_META ALL_USER ALWAYS ALLOW ANALYSE ANY + ACCESS ACCESS_INFO ACCESSID ACCESSKEY ACCESSTYPE ACCOUNT ACTION ACTIVE ADDDATE AFTER AGAINST AGGREGATE AI AI_SPLIT_DOCUMENT ALGORITHM ALL_META ALL_USER ALWAYS ALLOW ANALYSE ANY APPID APPROX_COUNT_DISTINCT APPROX_COUNT_DISTINCT_SYNOPSIS APPROX_COUNT_DISTINCT_SYNOPSIS_MERGE ARRAY ASCII ASIS AT ATTRIBUTE AUTHORS AUTO AUTOEXTEND_SIZE AUTO_INCREMENT AUTO_INCREMENT_MODE AUTO_INCREMENT_CACHE_SIZE AVG AVG_ROW_LENGTH ACTIVATE AVAILABILITY ARCHIVELOG ASYNCHRONOUS AUDIT ADMIN AUTO_REFRESH API_MODE APPROX APPROXIMATE ARRAY_AGG ARRAY_FILTER ARRAY_FIRST ARRAY_MAP ARRAY_SORTBY @@ -537,7 +537,7 @@ END_P SET_VAR DELIMITER %type skip_index_type opt_skip_index_type_list %type opt_rebuild_column_store %type vec_index_params vec_index_param vec_index_param_value opt_with_vector_index_parameters -%type json_table_expr rb_iterate_expr unnest_expr mock_jt_on_error_on_empty jt_column_list json_table_column_def +%type json_table_expr rb_iterate_expr unnest_expr mock_jt_on_error_on_empty jt_column_list json_table_column_def ai_split_document_expr %type json_table_ordinality_column_def json_table_exists_column_def json_table_value_column_def json_table_nested_column_def %type opt_value_on_empty_or_error_or_mismatch opt_on_mismatch %type table_values_clause table_values_clause_with_order_by_and_limit values_row_list row_value @@ -12974,6 +12974,10 @@ tbl_name { $$ = $1; } +| ai_split_document_expr +{ + $$ = $1; +} | '(' table_references ')' { $$ = $2; @@ -21024,6 +21028,45 @@ JSON_TABLE '(' simple_expr ',' literal mock_jt_on_error_on_empty COLUMNS '(' jt_ } ; +ai_split_document_expr: + AI_SPLIT_DOCUMENT '(' opt_expr_as_list ')' +{ + ParseNode *params = NULL; + if (NULL != $3) { + merge_nodes(params, result, T_EXPR_LIST, $3); + } + ParseNode *func_name = NULL; + make_name_node(func_name, result->malloc_pool_, "ai_split_document"); + ParseNode *func_expr = NULL; + malloc_non_terminal_node(func_expr, result->malloc_pool_, T_FUN_SYS, 2, func_name, params); + malloc_non_terminal_node($$, result->malloc_pool_, T_TABLE_COLLECTION_EXPRESSION, 2, func_expr, NULL); +} +| AI_SPLIT_DOCUMENT '(' opt_expr_as_list ')' relation_name +{ + ParseNode *params = NULL; + if (NULL != $3) { + merge_nodes(params, result, T_EXPR_LIST, $3); + } + ParseNode *func_name = NULL; + make_name_node(func_name, result->malloc_pool_, "ai_split_document"); + ParseNode *func_expr = NULL; + malloc_non_terminal_node(func_expr, result->malloc_pool_, T_FUN_SYS, 2, func_name, params); + malloc_non_terminal_node($$, result->malloc_pool_, T_TABLE_COLLECTION_EXPRESSION, 2, func_expr, $5); +} +| AI_SPLIT_DOCUMENT '(' opt_expr_as_list ')' AS relation_name +{ + ParseNode *params = NULL; + if (NULL != $3) { + merge_nodes(params, result, T_EXPR_LIST, $3); + } + ParseNode *func_name = NULL; + make_name_node(func_name, result->malloc_pool_, "ai_split_document"); + ParseNode *func_expr = NULL; + malloc_non_terminal_node(func_expr, result->malloc_pool_, T_FUN_SYS, 2, func_name, params); + malloc_non_terminal_node($$, result->malloc_pool_, T_TABLE_COLLECTION_EXPRESSION, 2, func_expr, $6); +} +; + mock_jt_on_error_on_empty: { ParseNode *emp_node = NULL; @@ -22004,6 +22047,7 @@ ACCESS_INFO | ADMIN | AFTER | AI +| AI_SPLIT_DOCUMENT | AGAINST | AGGREGATE | ALGORITHM diff --git a/src/sql/resolver/dml/ob_dml_resolver.cpp b/src/sql/resolver/dml/ob_dml_resolver.cpp index d18a7b6adc..948f6976cf 100755 --- a/src/sql/resolver/dml/ob_dml_resolver.cpp +++ b/src/sql/resolver/dml/ob_dml_resolver.cpp @@ -9236,8 +9236,40 @@ int ObDMLResolver::resolve_function_table_column_item_sys_func(const TableItem & } else if (!ObResolverUtils::is_expr_can_be_used_in_table_function(*table_expr)) { ret = OB_NOT_SUPPORTED; LOG_USER_ERROR(OB_NOT_SUPPORTED, "access rows from a non-nested table item"); + } else if (T_FUN_SYS_AI_SPLIT_DOCUMENT == table_expr->get_expr_type()) { + struct ColDef { const char *name; ObObjType type; }; + static const ColDef COLS[4] = { + {"CHUNK_ID", ObIntType}, + {"CHUNK_OFFSET", ObIntType}, + {"CHUNK_LENGTH", ObIntType}, + {"CHUNK_TEXT", ObVarcharType}, + }; + // varchar columns need a valid collation to formalize/deduce_type; use the + // connection collation (matches how the rest of the resolver types string + // columns). int columns are unaffected. + ObCollationType coll_connection = CS_TYPE_INVALID; + OZ (params_.session_info_->get_collation_connection(coll_connection)); + for (int64_t i = 0; OB_SUCC(ret) && i < 4; ++i) { + ColumnItem *c = NULL; + ObObjMeta meta; + meta.set_type(COLS[i].type); + if (ob_is_string_type(COLS[i].type)) { + meta.set_collation_type(coll_connection); + meta.set_collation_level(CS_LEVEL_IMPLICIT); + } + ObAccuracy accuracy; + ObString col_name(COLS[i].name); + if (NULL == (c = stmt->get_column_item(table_item.table_id_, col_name))) { + OZ (resolve_function_table_column_item(table_item, meta, accuracy, + col_name, OB_APP_MIN_COLUMN_ID + i, c)); + } + CK (OB_NOT_NULL(c)); + OZ (col_items.push_back(*c)); + } } else if (NULL != (col_item = stmt->get_column_item(table_item.table_id_, ObString("COLUMN_VALUE")))) { //exist, ignore resolve... + CK (OB_NOT_NULL(col_item)); + OZ (col_items.push_back(*col_item)); } else { OZ (resolve_function_table_column_item(table_item, table_expr->get_result_meta(), @@ -9245,9 +9277,9 @@ int ObDMLResolver::resolve_function_table_column_item_sys_func(const TableItem & ObString("COLUMN_VALUE"), OB_APP_MIN_COLUMN_ID, col_item)); + CK (OB_NOT_NULL(col_item)); + OZ (col_items.push_back(*col_item)); } - CK (OB_NOT_NULL(col_item)); - OZ (col_items.push_back(*col_item)); return ret; } diff --git a/src/sql/resolver/ob_resolver_utils.cpp b/src/sql/resolver/ob_resolver_utils.cpp index 0375eac179..8f31db5f0c 100644 --- a/src/sql/resolver/ob_resolver_utils.cpp +++ b/src/sql/resolver/ob_resolver_utils.cpp @@ -95,67 +95,78 @@ int ObResolverUtils::get_all_function_table_column_names(const TableItem &table_ CK (OB_NOT_NULL(package_guard)); CK (OB_LIKELY(table_item.is_function_table())); CK (OB_NOT_NULL(table_expr = table_item.function_table_expr_)); - CK (table_expr->get_udt_id() != OB_INVALID_ID); - - CK (OB_NOT_NULL(params.schema_checker_)); - OZ (ObResolverUtils::get_user_type( - params.allocator_, params.session_info_, params.sql_proxy_, - params.schema_checker_->get_schema_guard(), - *package_guard, - table_expr->get_udt_id(), user_type)); - CK (OB_NOT_NULL(user_type)); - if (OB_SUCC(ret) && !user_type->is_collection_type()) { - ret = OB_NOT_SUPPORTED; - LOG_WARN("function table get udf with return type not table type", - K(ret), K(user_type->is_collection_type())); - LOG_USER_ERROR(OB_NOT_SUPPORTED, "user define type is not collation type in function table"); - } - const ObCollectionType *coll_type = NULL; - CK (OB_NOT_NULL(coll_type = static_cast(user_type))); - if (OB_SUCC(ret) - && !coll_type->get_element_type().is_obj_type() - && !coll_type->get_element_type().is_record_type() - && !coll_type->get_element_type().is_collection_type() - && !(coll_type->get_element_type().is_opaque_type() - && coll_type->get_element_type().get_user_type_id() == T_OBJ_XML)) { - ret = OB_NOT_SUPPORTED; - LOG_WARN("not suppoert type in table function", K(ret), KPC(coll_type)); - ObString err; - err.write(coll_type->get_name().ptr(), coll_type->get_name().length()); - err.write(" collation type in table function\0", sizeof(" collation type in table function\0")); - LOG_USER_ERROR(OB_NOT_SUPPORTED, err.ptr()); - } - if (OB_SUCC(ret) && (coll_type->get_element_type().is_obj_type() - || coll_type->get_element_type().is_opaque_type() - || coll_type->get_element_type().is_collection_type())) { + if (T_FUN_SYS_AI_SPLIT_DOCUMENT == table_expr->get_expr_type()) { + // sys_func table function: no PL UDT; return the 4 fixed column names. + static const char *AI_SPLIT_COLS[4] = {"chunk_id", "chunk_offset", "chunk_length", "chunk_text"}; + for (int64_t i = 0; OB_SUCC(ret) && i < 4; ++i) { + ObString name(AI_SPLIT_COLS[i]); + OZ (column_names.push_back(name)); + } + } else if (T_FUN_SYS_GENERATOR == table_expr->get_expr_type()) { OZ (column_names.push_back(ObString("COLUMN_VALUE"))); - } - if (OB_SUCC(ret) && coll_type->get_element_type().is_record_type()) { - const ObRecordType *record_type = NULL; - const ObUserDefinedType *user_type = NULL; + } else { + CK (table_expr->get_udt_id() != OB_INVALID_ID); + CK (OB_NOT_NULL(params.schema_checker_)); OZ (ObResolverUtils::get_user_type( params.allocator_, params.session_info_, params.sql_proxy_, params.schema_checker_->get_schema_guard(), *package_guard, - coll_type->get_element_type().get_user_type_id(), user_type)); + table_expr->get_udt_id(), user_type)); CK (OB_NOT_NULL(user_type)); - CK (user_type->is_record_type()); - CK (OB_NOT_NULL(record_type = static_cast(user_type))); - for (int64_t i = 0; OB_SUCC(ret) && i < record_type->get_member_count(); ++i) { - ObString name; - const ObString *member_name = record_type->get_record_member_name(i); - CK (OB_NOT_NULL(member_name)); + if (OB_SUCC(ret) && !user_type->is_collection_type()) { + ret = OB_NOT_SUPPORTED; + LOG_WARN("function table get udf with return type not table type", + K(ret), K(user_type->is_collection_type())); + LOG_USER_ERROR(OB_NOT_SUPPORTED, "user define type is not collation type in function table"); + } + const ObCollectionType *coll_type = NULL; + CK (OB_NOT_NULL(coll_type = static_cast(user_type))); + if (OB_SUCC(ret) + && !coll_type->get_element_type().is_obj_type() + && !coll_type->get_element_type().is_record_type() + && !coll_type->get_element_type().is_collection_type() + && !(coll_type->get_element_type().is_opaque_type() + && coll_type->get_element_type().get_user_type_id() == T_OBJ_XML)) { + ret = OB_NOT_SUPPORTED; + LOG_WARN("not suppoert type in table function", K(ret), KPC(coll_type)); + ObString err; + err.write(coll_type->get_name().ptr(), coll_type->get_name().length()); + err.write(" collation type in table function\0", sizeof(" collation type in table function\0")); + LOG_USER_ERROR(OB_NOT_SUPPORTED, err.ptr()); + } + if (OB_SUCC(ret) && (coll_type->get_element_type().is_obj_type() + || coll_type->get_element_type().is_opaque_type() + || coll_type->get_element_type().is_collection_type())) { + OZ (column_names.push_back(ObString("COLUMN_VALUE"))); + } + if (OB_SUCC(ret) && coll_type->get_element_type().is_record_type()) { + const ObRecordType *record_type = NULL; + const ObUserDefinedType *user_type = NULL; + CK (OB_NOT_NULL(params.schema_checker_)); + OZ (ObResolverUtils::get_user_type( + params.allocator_, params.session_info_, params.sql_proxy_, + params.schema_checker_->get_schema_guard(), + *package_guard, + coll_type->get_element_type().get_user_type_id(), user_type)); + CK (OB_NOT_NULL(user_type)); + CK (user_type->is_record_type()); + CK (OB_NOT_NULL(record_type = static_cast(user_type))); + for (int64_t i = 0; OB_SUCC(ret) && i < record_type->get_member_count(); ++i) { + ObString name; + const ObString *member_name = record_type->get_record_member_name(i); + CK (OB_NOT_NULL(member_name)); - if (OB_FAIL(ret)) { - // do nothing - } else if (PL_TYPE_PACKAGE == user_type->get_type_from()) { - OZ (ob_write_string(*params.allocator_, *member_name, name)); - } else { - name = *member_name; - } + if (OB_FAIL(ret)) { + // do nothing + } else if (PL_TYPE_PACKAGE == user_type->get_type_from()) { + OZ (ob_write_string(*params.allocator_, *member_name, name)); + } else { + name = *member_name; + } - OZ (column_names.push_back(name)); + OZ (column_names.push_back(name)); + } } } return ret; @@ -2982,6 +2993,9 @@ bool ObResolverUtils::is_expr_can_be_used_in_table_function(const ObRawExpr &exp } else if (T_FUN_SYS_GENERATOR == expr.get_expr_type()) { // for generator(N) stream function bret = true; + } else if (T_FUN_SYS_AI_SPLIT_DOCUMENT == expr.get_expr_type()) { + // for ai_split_document table function + bret = true; } return bret; }