|
| 1 | +// Licensed to the Apache Software Foundation (ASF) under one |
| 2 | +// or more contributor license agreements. See the NOTICE file |
| 3 | +// distributed with this work for additional information |
| 4 | +// regarding copyright ownership. The ASF licenses this file |
| 5 | +// to you under the Apache License, Version 2.0 (the |
| 6 | +// "License"); you may not use this file except in compliance |
| 7 | +// with the License. You may obtain a copy of the License at |
| 8 | +// |
| 9 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | +// |
| 11 | +// Unless required by applicable law or agreed to in writing, |
| 12 | +// software distributed under the License is distributed on an |
| 13 | +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 14 | +// KIND, either express or implied. See the License for the |
| 15 | +// specific language governing permissions and limitations |
| 16 | +// under the License. |
| 17 | + |
| 18 | +#include "fluss.hpp" |
| 19 | + |
| 20 | +#include <iostream> |
| 21 | +#include <vector> |
| 22 | + |
| 23 | +static void check(const char* step, const fluss::Result& r) { |
| 24 | + if (!r.Ok()) { |
| 25 | + std::cerr << step << " failed: code=" << r.error_code |
| 26 | + << " msg=" << r.error_message << std::endl; |
| 27 | + std::exit(1); |
| 28 | + } |
| 29 | +} |
| 30 | + |
| 31 | +int main() { |
| 32 | + // 1) Connect |
| 33 | + fluss::Connection conn; |
| 34 | + check("connect", fluss::Connection::Connect("127.0.0.1:9123", conn)); |
| 35 | + |
| 36 | + // 2) Admin |
| 37 | + fluss::Admin admin; |
| 38 | + check("get_admin", conn.GetAdmin(admin)); |
| 39 | + |
| 40 | + // 3) Schema & descriptor |
| 41 | + auto schema = fluss::Schema::NewBuilder() |
| 42 | + .AddColumn("id", fluss::DataType::Int) |
| 43 | + .AddColumn("name", fluss::DataType::String) |
| 44 | + .AddColumn("score", fluss::DataType::Float) |
| 45 | + .AddColumn("age", fluss::DataType::Int) |
| 46 | + .Build(); |
| 47 | + |
| 48 | + auto descriptor = fluss::TableDescriptor::NewBuilder() |
| 49 | + .SetSchema(schema) |
| 50 | + .SetBucketCount(1) |
| 51 | + .SetProperty("table.log.arrow.compression.type", "NONE") |
| 52 | + .SetComment("cpp example table") |
| 53 | + .Build(); |
| 54 | + |
| 55 | + fluss::TablePath table_path("fluss", "sample_table_cpp_v1"); |
| 56 | + // ignore_if_exists=true to allow re-run |
| 57 | + check("create_table", admin.CreateTable(table_path, descriptor, true)); |
| 58 | + |
| 59 | + // 4) Get table |
| 60 | + fluss::Table table; |
| 61 | + check("get_table", conn.GetTable(table_path, table)); |
| 62 | + |
| 63 | + // 5) Writer |
| 64 | + fluss::AppendWriter writer; |
| 65 | + check("new_append_writer", table.NewAppendWriter(writer)); |
| 66 | + |
| 67 | + struct RowData { |
| 68 | + int id; |
| 69 | + const char* name; |
| 70 | + float score; |
| 71 | + int age; |
| 72 | + }; |
| 73 | + |
| 74 | + std::vector<RowData> rows = { |
| 75 | + {1, "Alice", 95.2f, 25}, |
| 76 | + {2, "Bob", 87.2f, 30}, |
| 77 | + {3, "Charlie", 92.1f, 35}, |
| 78 | + }; |
| 79 | + |
| 80 | + for (const auto& r : rows) { |
| 81 | + fluss::GenericRow row; |
| 82 | + row.SetInt32(0, r.id); |
| 83 | + row.SetString(1, r.name); |
| 84 | + row.SetFloat32(2, r.score); |
| 85 | + row.SetInt32(3, r.age); |
| 86 | + check("append", writer.Append(row)); |
| 87 | + } |
| 88 | + check("flush", writer.Flush()); |
| 89 | + std::cout << "Wrote " << rows.size() << " rows" << std::endl; |
| 90 | + |
| 91 | + // 6) Scan |
| 92 | + fluss::LogScanner scanner; |
| 93 | + check("new_log_scanner", table.NewLogScanner(scanner)); |
| 94 | + |
| 95 | + auto info = table.GetTableInfo(); |
| 96 | + int buckets = info.num_buckets; |
| 97 | + for (int b = 0; b < buckets; ++b) { |
| 98 | + check("subscribe", scanner.Subscribe(b, 0)); |
| 99 | + } |
| 100 | + |
| 101 | + fluss::ScanRecords records; |
| 102 | + check("poll", scanner.Poll(5000, records)); |
| 103 | + |
| 104 | + std::cout << "Scanned records: " << records.records.size() << std::endl; |
| 105 | + for (const auto& rec : records.records) { |
| 106 | + std::cout << " offset=" << rec.offset << " id=" << rec.row.fields[0].i32_val |
| 107 | + << " name=" << rec.row.fields[1].string_val |
| 108 | + << " score=" << rec.row.fields[2].f32_val << " age=" << rec.row.fields[3].i32_val |
| 109 | + << " ts=" << rec.timestamp << std::endl; |
| 110 | + } |
| 111 | + |
| 112 | + // 7) Project only id (0) and name (1) columns |
| 113 | + std::vector<size_t> projected_columns = {0, 1}; |
| 114 | + fluss::LogScanner projected_scanner; |
| 115 | + check("new_log_scanner_with_projection", |
| 116 | + table.NewLogScannerWithProjection(projected_columns, projected_scanner)); |
| 117 | + |
| 118 | + for (int b = 0; b < buckets; ++b) { |
| 119 | + check("subscribe_projected", projected_scanner.Subscribe(b, 0)); |
| 120 | + } |
| 121 | + |
| 122 | + fluss::ScanRecords projected_records; |
| 123 | + check("poll_projected", projected_scanner.Poll(5000, projected_records)); |
| 124 | + |
| 125 | + std::cout << "Projected records: " << projected_records.records.size() << std::endl; |
| 126 | + |
| 127 | + bool projection_verified = true; |
| 128 | + for (size_t i = 0; i < projected_records.records.size(); ++i) { |
| 129 | + const auto& rec = projected_records.records[i]; |
| 130 | + const auto& row = rec.row; |
| 131 | + |
| 132 | + if (row.fields.size() != projected_columns.size()) { |
| 133 | + std::cerr << "ERROR: Record " << i << " has " << row.fields.size() |
| 134 | + << " fields, expected " << projected_columns.size() << std::endl; |
| 135 | + projection_verified = false; |
| 136 | + continue; |
| 137 | + } |
| 138 | + |
| 139 | + // Verify field types match expected columns |
| 140 | + // Column 0 (id) should be Int32, Column 1 (name) should be String |
| 141 | + if (row.fields[0].type != fluss::DatumType::Int32) { |
| 142 | + std::cerr << "ERROR: Record " << i << " field 0 type mismatch, expected Int32" << std::endl; |
| 143 | + projection_verified = false; |
| 144 | + } |
| 145 | + if (row.fields[1].type != fluss::DatumType::String) { |
| 146 | + std::cerr << "ERROR: Record " << i << " field 1 type mismatch, expected String" << std::endl; |
| 147 | + projection_verified = false; |
| 148 | + } |
| 149 | + |
| 150 | + // Print projected data |
| 151 | + if (row.fields[0].type == fluss::DatumType::Int32 && |
| 152 | + row.fields[1].type == fluss::DatumType::String) { |
| 153 | + std::cout << " Record " << i << ": id=" << row.fields[0].i32_val |
| 154 | + << ", name=" << row.fields[1].string_val << std::endl; |
| 155 | + } |
| 156 | + } |
| 157 | + |
| 158 | + if (projection_verified) { |
| 159 | + std::cout << "Column pruning verification passed!" << std::endl; |
| 160 | + } else { |
| 161 | + std::cerr << "Column pruning verification failed!" << std::endl; |
| 162 | + std::exit(1); |
| 163 | + } |
| 164 | + |
| 165 | + return 0; |
| 166 | +} |
0 commit comments