Struct TableScan 

pub struct TableScan<'a> { /* private fields */ }

Implementations

impl<'a> TableScan<'a>

pub fn new( conn: &'a FlussConnection, table_info: TableInfo, metadata: Arc<Metadata>, ) -> Self

pub fn project(self, column_indices: &[usize]) -> Result<Self>

Projects the scan to only include specified columns by their indices.

Arguments

• column_indices - Zero-based indices of columns to include in the scan

Errors

Returns an error if column_indices is empty or if any column index is out of range.

Example

    let mut config = Config::default();
    config.bootstrap_servers = "127.0.0.1:9123".to_string();
    let conn = FlussConnection::new(config).await?;

    let table_descriptor = TableDescriptor::builder()
        .schema(
            Schema::builder()
                .column("col1", DataTypes::int())
                .column("col2", DataTypes::string())
                .column("col3", DataTypes::string())
                .column("col4", DataTypes::string())
            .build()?,
        ).build()?;
    let table_path = TablePath::new("fluss".to_owned(), "rust_test_long".to_owned());
    let admin = conn.get_admin()?;
    admin.create_table(&table_path, &table_descriptor, true)
        .await?;
    let table_info = admin.get_table_info(&table_path).await?;
    let table = conn.get_table(&table_path).await?;

    // Project columns by indices
    let scanner = table.new_scan().project(&[0, 2, 3])?.create_log_scanner()?;
    let scan_records = scanner.poll(Duration::from_secs(10)).await?;
    for record in scan_records {
        let row = record.row();
        println!(
            "{{{}, {}, {}}}@{}",
            row.get_int(0)?,
            row.get_string(2)?,
            row.get_string(3)?,
            record.offset()
        );
    }

pub fn project_by_name(self, column_names: &[&str]) -> Result<Self>

Projects the scan to only include specified columns by their names.

Arguments

• column_names - Names of columns to include in the scan

Errors

Returns an error if column_names is empty or if any column name is not found in the table schema.

Example

    let mut config = Config::default();
    config.bootstrap_servers = "127.0.0.1:9123".to_string();
    let conn = FlussConnection::new(config).await?;

    let table_descriptor = TableDescriptor::builder()
        .schema(
            Schema::builder()
                .column("col1", DataTypes::int())
                .column("col2", DataTypes::string())
                .column("col3", DataTypes::string())
            .build()?,
        ).build()?;
    let table_path = TablePath::new("fluss".to_owned(), "rust_test_long".to_owned());
    let admin = conn.get_admin()?;
    admin.create_table(&table_path, &table_descriptor, true)
        .await?;
    let table = conn.get_table(&table_path).await?;

    // Project columns by column names
    let scanner = table.new_scan().project_by_name(&["col1", "col3"])?.create_log_scanner()?;
    let scan_records = scanner.poll(Duration::from_secs(10)).await?;
    for record in scan_records {
        let row = record.row();
        println!(
            "{{{}, {}}}@{}",
            row.get_int(0)?,
            row.get_string(1)?,
            record.offset()
        );
    }

pub fn create_log_scanner(self) -> Result<LogScanner>

pub fn create_record_batch_log_scanner(self) -> Result<RecordBatchLogScanner>