Repository: apache/arrow-rs
Pull request #6354 (2024-09-04 09:25:49)

Add breaking change from #6043 to `CHANGELOG`
Adds #6043 to the list of breaking changes in the `CHANGELOG` for `53.0.0`.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 72b809019..20c13ffeb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -32,6 +32,7 @@
 - Upgrade protobuf definitions to flightsql 17.0 \(\#6133\) [\#6169](https://github.com/apache/arrow-rs/pull/6169) [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([alamb](https://github.com/alamb))
 - Remove automatic buffering in `ipc::reader::FileReader` for for consistent buffering [\#6132](https://github.com/apache/arrow-rs/pull/6132) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([V0ldek](https://github.com/V0ldek))
 - No longer write Parquet column metadata after column chunks \*and\* in the footer [\#6117](https://github.com/apache/arrow-rs/pull/6117) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Remove `impl<T: AsRef<[u8]>> From<T> for Buffer` that easily accidentally copies data [\#6043](https://github.com/apache/arrow-rs/pull/6043) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
 
 **Implemented enhancements:**
 
diff --git a/arrow-buffer/Cargo.toml b/arrow-buffer/Cargo.toml
index 8bc33b187..68bfe8ddf 100644
--- a/arrow-buffer/Cargo.toml
+++ b/arrow-buffer/Cargo.toml
@@ -44,10 +44,14 @@ rand = { version = "0.8", default-features = false, features = ["std", "std_rng"
 
 [build-dependencies]
 
+[[bench]]
+name = "bit_mask"
+harness = false
+
 [[bench]]
 name = "i256"
 harness = false
 
 [[bench]]
 name = "offset"
-harness = false
\ No newline at end of file
+harness = false
diff --git a/arrow-buffer/benches/bit_mask.rs b/arrow-buffer/benches/bit_mask.rs
new file mode 100644
index 000000000..6907e336a
--- /dev/null
+++ b/arrow-buffer/benches/bit_mask.rs
@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow_buffer::bit_mask::set_bits;
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("bit_mask");
+
+    for offset_write in [0, 5] {
+        for offset_read in [0, 5] {
+            for len in [1, 17, 65] {
+                for datum in [0u8, 0xADu8] {
+                    let x = (offset_write, offset_read, len, datum);
+                    group.bench_with_input(
+                        BenchmarkId::new(
+                            "set_bits",
+                            format!(
+                                "offset_write_{}_offset_read_{}_len_{}_datum_{}",
+                                x.0, x.1, x.2, x.3
+                            ),
+                        ),
+                        &x,
+                        |b, &x| {
+                            b.iter(|| {
+                                set_bits(
+                                    black_box(&mut [0u8; 9]),
+                                    black_box(&[x.3; 9]),
+                                    black_box(x.0),
+                                    black_box(x.1),
+                                    black_box(x.2),
+                                )
+                            });
+                        },
+                    );
+                }
+            }
+        }
+    }
+    group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);
diff --git a/arrow-buffer/src/buffer/immutable.rs b/arrow-buffer/src/buffer/immutable.rs
index c53ef18ba..7cd355221 100644
--- a/arrow-buffer/src/buffer/immutable.rs
+++ b/arrow-buffer/src/buffer/immutable.rs
@@ -382,6 +382,12 @@ impl<const N: usize> From<&[u8; N]> for Buffer {
     }
 }
 
+impl<T: ArrowNativeType> From<Vec<T>> for Buffer {
+    fn from(value: Vec<T>) -> Self {
+        Self::from_vec(value)
+    }
+}
+
 /// Creating a `Buffer` instance by storing the boolean values into the buffer
 impl FromIterator<bool> for Buffer {
     fn from_iter<I>(iter: I) -> Self
diff --git a/arrow-flight/src/bin/flight_sql_client.rs b/arrow-flight/src/bin/flight_sql_client.rs
index 296efc1c3..c334b95a9 100644
--- a/arrow-flight/src/bin/flight_sql_client.rs
+++ b/arrow-flight/src/bin/flight_sql_client.rs
@@ -20,7 +20,10 @@ use std::{sync::Arc, time::Duration};
 use anyhow::{bail, Context, Result};
 use arrow_array::{ArrayRef, Datum, RecordBatch, StringArray};
 use arrow_cast::{cast_with_options, pretty::pretty_format_batches, CastOptions};
-use arrow_flight::{sql::client::FlightSqlServiceClient, FlightInfo};
+use arrow_flight::{
+    sql::{client::FlightSqlServiceClient, CommandGetDbSchemas, CommandGetTables},
+    FlightInfo,
+};
 use arrow_schema::Schema;
 use clap::{Parser, Subcommand};
 use futures::TryStreamExt;
@@ -111,6 +114,51 @@ struct Args {
 /// Different available commands.
 #[derive(Debug, Subcommand)]
 enum Command {
+    /// Get catalogs.
+    Catalogs,
+    /// Get db schemas for a catalog.
+    DbSchemas {
+        /// Name of a catalog.
+        ///
+        /// Required.
+        catalog: String,
+        /// Specifies a filter pattern for schemas to search for.
+        /// When no schema_filter is provided, the pattern will not be used to narrow the search.
+        /// In the pattern string, two special characters can be used to denote matching rules:
+        ///     - "%" means to match any substring with 0 or more characters.
+        ///     - "_" means to match any one character.
+        #[clap(short, long)]
+        db_schema_filter: Option<String>,
+    },
+    /// Get tables for a catalog.
+    Tables {
+        /// Name of a catalog.
+        ///
+        /// Required.
+        catalog: String,
+        /// Specifies a filter pattern for schemas to search for.
+        /// When no schema_filter is provided, the pattern will not be used to narrow the search.
+        /// In the pattern string, two special characters can be used to denote matching rules:
+        ///     - "%" means to match any substring with 0 or more characters.
+        ///     - "_" means to match any one character.
+        #[clap(short, long)]
+        db_schema_filter: Option<String>,
+        /// Specifies a filter pattern for tables to search for.
+        /// When no table_filter is provided, all tables matching other filters are searched.
+        /// In the pattern string, two special characters can be used to denote matching rules:
+        ///     - "%" means to match any substring with 0 or more characters.
+        ///     - "_" means to match any one character.
+        #[clap(short, long)]
+        table_filter: Option<String>,
+        /// Specifies a filter of table types which must match.
+        /// The table types depend on vendor/implementation. It is usually used to separate tables from views or system tables.
+        /// TABLE, VIEW, and SYSTEM TABLE are commonly supported.
+        #[clap(long)]
+        table_types: Vec<String>,
+    },
+    /// Get table types.
+    TableTypes,
+
     /// Execute given statement.
     StatementQuery {
         /// SQL query.
@@ -150,6 +198,36 @@ async fn main() -> Result<()> {
         .context("setup client")?;
 
     let flight_info = match args.cmd {
+        Command::Catalogs => client.get_catalogs().await.context("get catalogs")?,
+        Command::DbSchemas {
+            catalog,
+            db_schema_filter,
+        } => client
+            .get_db_schemas(CommandGetDbSchemas {
+                catalog: Some(catalog),
+                db_schema_filter_pattern: db_schema_filter,
+            })
+            .await
+            .context("get db schemas")?,
+        Command::Tables {
+            catalog,
+            db_schema_filter,
+            table_filter,
+            table_types,
+        } => client
+            .get_tables(CommandGetTables {
+                catalog: Some(catalog),
+                db_schema_filter_pattern: db_schema_filter,
+                table_name_filter_pattern: table_filter,
+                table_types,
+                // Schema is returned as ipc encoded bytes.
+                // We do not support returning the schema as there is no trivial mechanism
+                // to display the information to the user.
+                include_schema: false,
+            })
+            .await
+            .context("get tables")?,
+        Command::TableTypes => client.get_table_types().await.context("get table types")?,
         Command::StatementQuery { query } => client
             .execute(query, None)
             .await
diff --git a/arrow-flight/src/sql/metadata/mod.rs b/arrow-flight/src/sql/metadata/mod.rs
index 1e9881ffa..fd71149a3 100644
--- a/arrow-flight/src/sql/metadata/mod.rs
+++ b/arrow-flight/src/sql/metadata/mod.rs
@@ -33,6 +33,7 @@
 mod catalogs;
 mod db_schemas;
 mod sql_info;
+mod table_types;
 mod tables;
 mod xdbc_info;
 
diff --git a/arrow-flight/src/sql/metadata/table_types.rs b/arrow-flight/src/sql/metadata/table_types.rs
new file mode 100644
index 000000000..54cfe6fe2
--- /dev/null
+++ b/arrow-flight/src/sql/metadata/table_types.rs
@@ -0,0 +1,158 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! [`GetTableTypesBuilder`] for building responses to [`CommandGetTableTypes`] queries.
+//!
+//! [`CommandGetTableTypes`]: crate::sql::CommandGetTableTypes
+
+use std::sync::Arc;
+
+use arrow_array::{builder::StringBuilder, ArrayRef, RecordBatch};
+use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use arrow_select::take::take;
+use once_cell::sync::Lazy;
+
+use crate::error::*;
+use crate::sql::CommandGetTableTypes;
+
+use super::lexsort_to_indices;
+
+/// A builder for a [`CommandGetTableTypes`] response.
+///
+/// Builds rows like this:
+///
+/// * table_type: utf8,
+#[derive(Default)]
+pub struct GetTableTypesBuilder {
+    // array builder for table types
+    table_type: StringBuilder,
+}
+
+impl CommandGetTableTypes {
+    /// Create a builder suitable for constructing a response
+    pub fn into_builder(self) -> GetTableTypesBuilder {
+        self.into()
+    }
+}
+
+impl From<CommandGetTableTypes> for GetTableTypesBuilder {
+    fn from(_value: CommandGetTableTypes) -> Self {
+        Self::new()
+    }
+}
+
+impl GetTableTypesBuilder {
+    /// Create a new instance of [`GetTableTypesBuilder`]
+    pub fn new() -> Self {
+        Self {
+            table_type: StringBuilder::new(),
+        }
+    }
+
+    /// Append a row
+    pub fn append(&mut self, table_type: impl AsRef<str>) {
+        self.table_type.append_value(table_type);
+    }
+
+    /// builds a `RecordBatch` with the correct schema for a `CommandGetTableTypes` response
+    pub fn build(self) -> Result<RecordBatch> {
+        let schema = self.schema();
+        let Self { mut table_type } = self;
+
+        // Make the arrays
+        let table_type = table_type.finish();
+
+        let batch = RecordBatch::try_new(schema, vec![Arc::new(table_type) as ArrayRef])?;
+
+        // Order filtered results by table_type
+        let indices = lexsort_to_indices(batch.columns());
+        let columns = batch
+            .columns()
+            .iter()
+            .map(|c| take(c, &indices, None))
+            .collect::<std::result::Result<Vec<_>, _>>()?;
+
+        Ok(RecordBatch::try_new(batch.schema(), columns)?)
+    }
+
+    /// Return the schema of the RecordBatch that will be returned
+    /// from [`CommandGetTableTypes`]
+    pub fn schema(&self) -> SchemaRef {
+        get_table_types_schema()
+    }
+}
+
+fn get_table_types_schema() -> SchemaRef {
+    Arc::clone(&GET_TABLE_TYPES_SCHEMA)
+}
+
+/// The schema for [`CommandGetTableTypes`].
+static GET_TABLE_TYPES_SCHEMA: Lazy<SchemaRef> = Lazy::new(|| {
+    Arc::new(Schema::new(vec![Field::new(
+        "table_type",
+        DataType::Utf8,
+        false,
+    )]))
+});
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow_array::StringArray;
+
+    fn get_ref_batch() -> RecordBatch {
+        RecordBatch::try_new(
+            get_table_types_schema(),
+            vec![Arc::new(StringArray::from(vec![
+                "a_table_type",
+                "b_table_type",
+                "c_table_type",
+                "d_table_type",
+            ])) as ArrayRef],
+        )
+        .unwrap()
+    }
+
+    #[test]
+    fn test_table_types_are_sorted() {
+        let ref_batch = get_ref_batch();
+
+        let mut builder = GetTableTypesBuilder::new();
+        builder.append("b_table_type");
+        builder.append("a_table_type");
+        builder.append("d_table_type");
+        builder.append("c_table_type");
+        let schema_batch = builder.build().unwrap();
+
+        assert_eq!(schema_batch, ref_batch)
+    }
+
+    #[test]
+    fn test_builder_from_query() {
+        let ref_batch = get_ref_batch();
+        let query = CommandGetTableTypes {};
+
+        let mut builder = query.into_builder();
+        builder.append("a_table_type");
+        builder.append("b_table_type");
+        builder.append("c_table_type");
+        builder.append("d_table_type");
+        let schema_batch = builder.build().unwrap();
+
+        assert_eq!(schema_batch, ref_batch)
+    }
+}
diff --git a/conbench/.flake8 b/conbench/.flake8
deleted file mode 100644
index e44b81084..000000000
--- a/conbench/.flake8
+++ /dev/null
@@ -1,2 +0,0 @@
-[flake8]
-ignore = E501
diff --git a/conbench/.gitignore b/conbench/.gitignore
deleted file mode 100755
index aa44ee2ad..000000000
--- a/conbench/.gitignore
+++ /dev/null
@@ -1,130 +0,0 @@
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[cod]
-*$py.class
-
-# C extensions
-*.so
-
-# Distribution / packaging
-.Python
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-pip-wheel-metadata/
-share/python-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
-MANIFEST
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.nox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*.cover
-*.py,cover
-.hypothesis/
-.pytest_cache/
-
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
-*.log
-local_settings.py
-db.sqlite3
-db.sqlite3-journal
-
-# Flask stuff:
-instance/
-.webassets-cache
-
-# Scrapy stuff:
-.scrapy
-
-# Sphinx documentation
-docs/_build/
-
-# PyBuilder
-target/
-
-# Jupyter Notebook
-.ipynb_checkpoints
-
-# IPython
-profile_default/
-ipython_config.py
-
-# pyenv
-.python-version
-
-# pipenv
-#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
-#   However, in case of collaboration, if having platform-specific dependencies or dependencies
-#   having no cross-platform support, pipenv may install dependencies that don't work, or not
-#   install all needed dependencies.
-#Pipfile.lock
-
-# PEP 582; used by e.g. github.com/David-OConnor/pyflow
-__pypackages__/
-
-# Celery stuff
-celerybeat-schedule
-celerybeat.pid
-
-# SageMath parsed files
-*.sage.py
-
-# Environments
-.env
-.venv
-env/
-venv/
-ENV/
-env.bak/
-venv.bak/
-
-# Spyder project settings
-.spyderproject
-.spyproject
-
-# Rope project settings
-.ropeproject
-
-# mkdocs documentation
-/site
-
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
-# Pyre type checker
-.pyre/
-
diff --git a/conbench/.isort.cfg b/conbench/.isort.cfg
deleted file mode 100644
index f238bf7ea..000000000
--- a/conbench/.isort.cfg
+++ /dev/null
@@ -1,2 +0,0 @@
-[settings]
-profile = black
diff --git a/conbench/README.md b/conbench/README.md
deleted file mode 100644
index 8c7f38cb3..000000000
--- a/conbench/README.md
+++ /dev/null
@@ -1,251 +0,0 @@
-<!---
-  Licensed to the Apache Software Foundation (ASF) under one
-  or more contributor license agreements.  See the NOTICE file
-  distributed with this work for additional information
-  regarding copyright ownership.  The ASF licenses this file
-  to you under the Apache License, Version 2.0 (the
-  "License"); you may not use this file except in compliance
-  with the License.  You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-  Unless required by applicable law or agreed to in writing,
-  software distributed under the License is distributed on an
-  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-  KIND, either express or implied.  See the License for the
-  specific language governing permissions and limitations
-  under the License.
--->
-
-# Arrow Rust + Conbench Integration
-
-
-## Quick start
-
-```
-$ cd ~/arrow-rs/conbench/
-$ conda create -y -n conbench python=3.9
-$ conda activate conbench
-(conbench) $ pip install -r requirements.txt
-(conbench) $ conbench arrow-rs
-```
-
-## Example output
-
-```
-{
-    "batch_id": "b68c559358cc43a3aab02d893d2693f4",
-    "context": {
-        "benchmark_language": "Rust"
-    },
-    "github": {
-        "commit": "ca33a0a50494f95840ade2e9509c3c3d4df35249",
-        "repository": "https://github.com/dianaclarke/arrow-rs"
-    },
-    "info": {},
-    "machine_info": {
-        "architecture_name": "x86_64",
-        "cpu_core_count": "8",
-        "cpu_frequency_max_hz": "2400000000",
-        "cpu_l1d_cache_bytes": "65536",
-        "cpu_l1i_cache_bytes": "131072",
-        "cpu_l2_cache_bytes": "4194304",
-        "cpu_l3_cache_bytes": "0",
-        "cpu_model_name": "Apple M1",
-        "cpu_thread_count": "8",
-        "gpu_count": "0",
-        "gpu_product_names": [],
-        "kernel_name": "20.6.0",
-        "memory_bytes": "17179869184",
-        "name": "diana",
-        "os_name": "macOS",
-        "os_version": "10.16"
-    },
-    "run_id": "08353595bde147fb9deebdb4facd019a",
-    "stats": {
-        "data": [
-            "0.000287",
-            "0.000286",
-            "0.000285",
-            "0.000281",
-            "0.000282",
-            "0.000277",
-            "0.000277",
-            "0.000286",
-            "0.000279",
-            "0.000282",
-            "0.000277",
-            "0.000277",
-            "0.000282",
-            "0.000276",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000284",
-            "0.000288",
-            "0.000278",
-            "0.000276",
-            "0.000278",
-            "0.000275",
-            "0.000275",
-            "0.000275",
-            "0.000275",
-            "0.000281",
-            "0.000284",
-            "0.000277",
-            "0.000277",
-            "0.000278",
-            "0.000282",
-            "0.000281",
-            "0.000284",
-            "0.000282",
-            "0.000279",
-            "0.000280",
-            "0.000281",
-            "0.000281",
-            "0.000286",
-            "0.000278",
-            "0.000278",
-            "0.000281",
-            "0.000276",
-            "0.000284",
-            "0.000281",
-            "0.000276",
-            "0.000276",
-            "0.000279",
-            "0.000283",
-            "0.000282",
-            "0.000278",
-            "0.000281",
-            "0.000284",
-            "0.000279",
-            "0.000276",
-            "0.000278",
-            "0.000283",
-            "0.000282",
-            "0.000276",
-            "0.000281",
-            "0.000279",
-            "0.000276",
-            "0.000277",
-            "0.000283",
-            "0.000279",
-            "0.000281",
-            "0.000283",
-            "0.000279",
-            "0.000282",
-            "0.000283",
-            "0.000278",
-            "0.000281",
-            "0.000282",
-            "0.000278",
-            "0.000276",
-            "0.000281",
-            "0.000278",
-            "0.000276",
-            "0.000282",
-            "0.000281",
-            "0.000282",
-            "0.000280",
-            "0.000281",
-            "0.000282",
-            "0.000280",
-            "0.000282",
-            "0.000280",
-            "0.000280",
-            "0.000282",
-            "0.000278",
-            "0.000284",
-            "0.000290",
-            "0.000282",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000278"
-        ],
-        "iqr": "0.000004",
-        "iterations": 100,
-        "max": "0.000290",
-        "mean": "0.000280",
-        "median": "0.000281",
-        "min": "0.000275",
-        "q1": "0.000278",
-        "q3": "0.000282",
-        "stdev": "0.000003",
-        "time_unit": "s",
-        "times": [],
-        "unit": "s"
-    },
-    "tags": {
-        "name": "nlike_utf8 scalar starts with",
-        "suite": "nlike_utf8 scalar starts with"
-    },
-    "timestamp": "2022-02-09T02:33:26.792404+00:00"
-}
-```
-
-## Debug with test benchmark
-
-```
-(conbench) $ cd ~/arrow-rs/conbench/
-(conbench) $ conbench test --iterations=3
-
-Benchmark result:
-{
-    "batch_id": "f4235d547e9d4f94925b54692e625d7d",
-    "context": {
-        "benchmark_language": "Python"
-    },
-    "github": {
-        "commit": "35e16be01e680e9381b2d1393c2e3f8e7acb7b13",
-        "repository": "https://github.com/dianaclarke/arrow-rs"
-    },
-    "info": {
-        "benchmark_language_version": "Python 3.9.7"
-    },
-    "machine_info": {
-        "architecture_name": "x86_64",
-        "cpu_core_count": "8",
-        "cpu_frequency_max_hz": "2400000000",
-        "cpu_l1d_cache_bytes": "65536",
-        "cpu_l1i_cache_bytes": "131072",
-        "cpu_l2_cache_bytes": "4194304",
-        "cpu_l3_cache_bytes": "0",
-        "cpu_model_name": "Apple M1",
-        "cpu_thread_count": "8",
-        "gpu_count": "0",
-        "gpu_product_names": [],
-        "kernel_name": "20.6.0",
-        "memory_bytes": "17179869184",
-        "name": "diana",
-        "os_name": "macOS",
-        "os_version": "10.16"
-    },
-    "run_id": "b2ca0d581cf14f21936276ff7ca5a940",
-    "stats": {
-        "data": [
-            "0.000002",
-            "0.000001",
-            "0.000001"
-        ],
-        "iqr": "0.000001",
-        "iterations": 3,
-        "max": "0.000002",
-        "mean": "0.000001",
-        "median": "0.000001",
-        "min": "0.000001",
-        "q1": "0.000001",
-        "q3": "0.000002",
-        "stdev": "0.000001",
-        "time_unit": "s",
-        "times": [],
-        "unit": "s"
-    },
-    "tags": {
-        "name": "test"
-    },
-    "timestamp": "2022-02-09T01:55:52.250727+00:00"
-}
-```
diff --git a/conbench/_criterion.py b/conbench/_criterion.py
deleted file mode 100644
index 168a1b9b6..000000000
--- a/conbench/_criterion.py
+++ /dev/null
@@ -1,98 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-import collections
-import csv
-import os
-import pathlib
-import subprocess
-
-import conbench.runner
-from conbench.machine_info import github_info
-
-
-def _result_in_seconds(row):
-    # sample_measured_value - The value of the measurement for this sample.
-    # Note that this is the measured value for the whole sample, not the
-    # time-per-iteration To calculate the time-per-iteration, use
-    # sample_measured_value/iteration_count
-    # -- https://bheisler.github.io/criterion.rs/book/user_guide/csv_output.html
-    count = int(row["iteration_count"])
-    sample = float(row["sample_measured_value"])
-    return sample / count / 10**9
-
-
-def _parse_benchmark_group(row):
-    parts = row["group"].split(",")
-    if len(parts) > 1:
-        suite, name = parts[0], ",".join(parts[1:])
-    else:
-        suite, name = row["group"], row["group"]
-    return suite.strip(), name.strip()
-
-
-def _read_results(src_dir):
-    results = collections.defaultdict(lambda: collections.defaultdict(list))
-    path = pathlib.Path(os.path.join(src_dir, "target", "criterion"))
-    for path in list(path.glob("**/new/raw.csv")):
-        with open(path) as csv_file:
-            reader = csv.DictReader(csv_file)
-            for row in reader:
-                suite, name = _parse_benchmark_group(row)
-                results[suite][name].append(_result_in_seconds(row))
-    return results
-
-
-def _execute_command(command):
-    try:
-        print(command)
-        result = subprocess.run(command, capture_output=True, check=True)
-    except subprocess.CalledProcessError as e:
-        print(e.stderr.decode("utf-8"))
-        raise e
-    return result.stdout.decode("utf-8"), result.stderr.decode("utf-8")
-
-
-class CriterionBenchmark(conbench.runner.Benchmark):
-    external = True
-
-    def run(self, **kwargs):
-        src_dir = os.path.join(os.getcwd(), "..")
-        self._cargo_bench(src_dir)
-        results = _read_results(src_dir)
-        for suite in results:
-            self.conbench.mark_new_batch()
-            for name, data in results[suite].items():
-                yield self._record_result(suite, name, data, kwargs)
-
-    def _cargo_bench(self, src_dir):
-        os.chdir(src_dir)
-        _execute_command(["cargo", "bench"])
-
-    def _record_result(self, suite, name, data, options):
-        tags = {"suite": suite}
-        result = {"data": data, "unit": "s"}
-        context = {"benchmark_language": "Rust"}
-        github = github_info()
-        return self.conbench.record(
-            result,
-            name,
-            tags=tags,
-            context=context,
-            github=github,
-            options=options,
-        )
diff --git a/conbench/benchmarks.json b/conbench/benchmarks.json
deleted file mode 100644
index 3e1f33a7d..000000000
--- a/conbench/benchmarks.json
+++ /dev/null
@@ -1,8 +0,0 @@
-[
-  {
-    "command": "arrow-rs",
-    "flags": {
-      "language": "Rust"
-    }
-  }
-]
diff --git a/conbench/benchmarks.py b/conbench/benchmarks.py
deleted file mode 100644
index bc4c1796b..000000000
--- a/conbench/benchmarks.py
+++ /dev/null
@@ -1,41 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-import conbench.runner
-
-import _criterion
-
-
-@conbench.runner.register_benchmark
-class TestBenchmark(conbench.runner.Benchmark):
-    name = "test"
-
-    def run(self, **kwargs):
-        yield self.conbench.benchmark(
-            self._f(),
-            self.name,
-            options=kwargs,
-        )
-
-    def _f(self):
-        return lambda: 1 + 1
-
-
-@conbench.runner.register_benchmark
-class CargoBenchmarks(_criterion.CriterionBenchmark):
-    name = "arrow-rs"
-    description = "Run Arrow Rust micro benchmarks."
diff --git a/conbench/requirements.txt b/conbench/requirements.txt
deleted file mode 100644
index a877c7b44..000000000
--- a/conbench/requirements.txt
+++ /dev/null
@@ -1,1 +0,0 @@
-conbench
diff --git a/dev/release/rat_exclude_files.txt b/dev/release/rat_exclude_files.txt
index 6f9d1b5f3..d08a0ea8c 100644
--- a/dev/release/rat_exclude_files.txt
+++ b/dev/release/rat_exclude_files.txt
@@ -14,11 +14,6 @@ parquet_derive/test/dependency/*
 Cargo.lock
 filtered_rat.txt
 rat.txt
-conbench/benchmarks.json
-conbench/requirements.txt
-conbench/requirements-test.txt
-conbench/.flake8
-conbench/.isort.cfg
 # auto-generated
 arrow-flight/src/arrow.flight.protocol.rs
 arrow-flight/src/sql/arrow.flight.protocol.sql.rs
diff --git a/object_store/src/delimited.rs b/object_store/src/delimited.rs
index 4f25c9d6d..96f88bf41 100644
--- a/object_store/src/delimited.rs
+++ b/object_store/src/delimited.rs
@@ -126,7 +126,7 @@ impl LineDelimiter {
     fn finish(&mut self) -> Result<bool> {
         if !self.remainder.is_empty() {
             ensure!(!self.is_quote, UnterminatedStringSnafu);
-            ensure!(!self.is_quote, TrailingEscapeSnafu);
+            ensure!(!self.is_escape, TrailingEscapeSnafu);
 
             self.complete
                 .push_back(Bytes::from(std::mem::take(&mut self.remainder)))
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 4b43f0cde..8820983b2 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -911,7 +911,7 @@ pub struct ObjectMeta {
 }
 
 /// Options for a get request, such as range
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct GetOptions {
     /// Request will succeed if the `ObjectMeta::e_tag` matches
     /// otherwise returning [`Error::Precondition`]


Test patch:
diff --git a/arrow-flight/tests/flight_sql_client_cli.rs b/arrow-flight/tests/flight_sql_client_cli.rs
index 168015d07..6e1f6142c 100644
--- a/arrow-flight/tests/flight_sql_client_cli.rs
+++ b/arrow-flight/tests/flight_sql_client_cli.rs
@@ -23,10 +23,12 @@ use crate::common::fixture::TestFixture;
 use arrow_array::{ArrayRef, Int64Array, RecordBatch, StringArray};
 use arrow_flight::{
     decode::FlightRecordBatchStream,
+    encode::FlightDataEncoderBuilder,
     flight_service_server::{FlightService, FlightServiceServer},
     sql::{
         server::{FlightSqlService, PeekableFlightDataStream},
         ActionCreatePreparedStatementRequest, ActionCreatePreparedStatementResult, Any,
+        CommandGetCatalogs, CommandGetDbSchemas, CommandGetTableTypes, CommandGetTables,
         CommandPreparedStatementQuery, CommandStatementQuery, DoPutPreparedStatementResult,
         ProstMessageExt, SqlInfo,
     },
@@ -85,6 +87,205 @@ async fn test_simple() {
     );
 }
 
+#[tokio::test]
+async fn test_get_catalogs() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("catalogs")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+\
+        \n| catalog_name |\
+        \n+--------------+\
+        \n| catalog_a    |\
+        \n| catalog_b    |\
+        \n+--------------+",
+    );
+}
+
+#[tokio::test]
+async fn test_get_db_schemas() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("db-schemas")
+            .arg("catalog_a")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+----------------+\
+        \n| catalog_name | db_schema_name |\
+        \n+--------------+----------------+\
+        \n| catalog_a    | schema_1       |\
+        \n| catalog_a    | schema_2       |\
+        \n+--------------+----------------+",
+    );
+}
+
+#[tokio::test]
+async fn test_get_tables() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("tables")
+            .arg("catalog_a")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+----------------+------------+------------+\
+        \n| catalog_name | db_schema_name | table_name | table_type |\
+        \n+--------------+----------------+------------+------------+\
+        \n| catalog_a    | schema_1       | table_1    | TABLE      |\
+        \n| catalog_a    | schema_2       | table_2    | VIEW       |\
+        \n+--------------+----------------+------------+------------+",
+    );
+}
+#[tokio::test]
+async fn test_get_tables_db_filter() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("tables")
+            .arg("catalog_a")
+            .arg("--db-schema-filter")
+            .arg("schema_2")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+----------------+------------+------------+\
+        \n| catalog_name | db_schema_name | table_name | table_type |\
+        \n+--------------+----------------+------------+------------+\
+        \n| catalog_a    | schema_2       | table_2    | VIEW       |\
+        \n+--------------+----------------+------------+------------+",
+    );
+}
+
+#[tokio::test]
+async fn test_get_tables_types() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("table-types")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+\
+        \n| table_type   |\
+        \n+--------------+\
+        \n| SYSTEM_TABLE |\
+        \n| TABLE        |\
+        \n| VIEW         |\
+        \n+--------------+",
+    );
+}
+
 const PREPARED_QUERY: &str = "SELECT * FROM table WHERE field = $1";
 const PREPARED_STATEMENT_HANDLE: &str = "prepared_statement_handle";
 const UPDATED_PREPARED_STATEMENT_HANDLE: &str = "updated_prepared_statement_handle";
@@ -278,6 +479,84 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(resp)
     }
 
+    async fn get_flight_info_catalogs(
+        &self,
+        query: CommandGetCatalogs,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
+    async fn get_flight_info_schemas(
+        &self,
+        query: CommandGetDbSchemas,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
+
+    async fn get_flight_info_tables(
+        &self,
+        query: CommandGetTables,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
+
+    async fn get_flight_info_table_types(
+        &self,
+        query: CommandGetTableTypes,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
     async fn get_flight_info_statement(
         &self,
         query: CommandStatementQuery,
@@ -309,6 +588,109 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(resp)
     }
 
+    async fn do_get_catalogs(
+        &self,
+        query: CommandGetCatalogs,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for catalog_name in ["catalog_a", "catalog_b"] {
+            builder.append(catalog_name);
+        }
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    async fn do_get_schemas(
+        &self,
+        query: CommandGetDbSchemas,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for (catalog_name, schema_name) in [
+            ("catalog_a", "schema_1"),
+            ("catalog_a", "schema_2"),
+            ("catalog_b", "schema_3"),
+        ] {
+            builder.append(catalog_name, schema_name);
+        }
+
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    async fn do_get_tables(
+        &self,
+        query: CommandGetTables,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for (catalog_name, schema_name, table_name, table_type, schema) in [
+            (
+                "catalog_a",
+                "schema_1",
+                "table_1",
+                "TABLE",
+                Arc::new(Schema::empty()),
+            ),
+            (
+                "catalog_a",
+                "schema_2",
+                "table_2",
+                "VIEW",
+                Arc::new(Schema::empty()),
+            ),
+            (
+                "catalog_b",
+                "schema_3",
+                "table_3",
+                "TABLE",
+                Arc::new(Schema::empty()),
+            ),
+        ] {
+            builder
+                .append(catalog_name, schema_name, table_name, table_type, &schema)
+                .unwrap();
+        }
+
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    async fn do_get_table_types(
+        &self,
+        query: CommandGetTableTypes,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for table_type in ["TABLE", "VIEW", "SYSTEM_TABLE"] {
+            builder.append(table_type);
+        }
+
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
     async fn do_put_prepared_statement_query(
         &self,
         _query: CommandPreparedStatementQuery,
diff --git a/conbench/requirements-test.txt b/conbench/requirements-test.txt
deleted file mode 100644
index 5e5647acd..000000000
--- a/conbench/requirements-test.txt
+++ /dev/null
@@ -1,3 +0,0 @@
-black
-flake8
-isort