[dnr] copy from demo

WORKSPACE

@@ -486,6 +486,17 @@ crates_repository(
             # Note: This is a target we add from the additive build file above.
             deps = [":zlib"],
         )],
+        # TODO(parkmycar): Refactor this to build the version of zlib from the `bzip2-sys` crate.
+        "bzip2-sys": [crate.annotation(
+            gen_build_script = False,
+            deps = ["@bzip2"],
+        )],
+        "lzma-sys": [crate.annotation(
+            additive_build_file = "@//misc/bazel/c_deps:rust-sys/BUILD.lzma-sys.bazel",
+            gen_build_script = False,
+            # Note: This is a target we add from the additive build file above.
+            deps = [":xz"],
+        )],
         "openssl-sys": [crate.annotation(
             build_script_data = [
                 "@openssl//:openssl_lib",

misc/bazel/c_deps/rust-sys/BUILD.lzma-sys.bazel

@@ -0,0 +1,68 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Licensed 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 in the LICENSE file at the
+# root of this repository, or online 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.
+
+"""Builds xz from the lzma-sys crate."""
+
+cc_library(
+    name = "xz",
+    srcs = glob(
+        include = [
+            "xz-5.2/src/liblzma/api/**/*.c",
+            "xz-5.2/src/liblzma/lzma/**/*.c",
+            "xz-5.2/src/liblzma/lzma/**/*.h",
+            "xz-5.2/src/liblzma/lz/**/*.c",
+            "xz-5.2/src/liblzma/lz/**/*.h",
+            "xz-5.2/src/liblzma/check/**/*.c",
+            "xz-5.2/src/liblzma/check/**/*.h",
+            "xz-5.2/src/liblzma/simple/**/*.c",
+            "xz-5.2/src/liblzma/simple/**/*.h",
+            "xz-5.2/src/liblzma/delta/**/*.c",
+            "xz-5.2/src/liblzma/delta/**/*.h",
+            "xz-5.2/src/liblzma/common/**/*.c",
+            "xz-5.2/src/liblzma/common/**/*.h",
+            "xz-5.2/src/liblzma/rangecoder/**/*.c",
+            "xz-5.2/src/liblzma/rangecoder/**/*.h",
+            "xz-5.2/src/common/**/*.h",
+        ],
+        exclude = [
+            "**/*crc32_small*",
+            "**/*crc64_small*",
+            "**/*tablegen*",
+        ],
+    ) + [
+        "config.h",
+        "xz-5.2/src/common/tuklib_cpucores.c",
+        "xz-5.2/src/common/tuklib_physmem.c",
+    ],
+    hdrs = glob(["xz-5.2/src/liblzma/api/**/*.h"]),
+    copts = [
+        "-std=c99",
+        "-pthread",
+    ],
+    includes = [
+        "xz-5.2/src/common",
+        "xz-5.2/src/liblzma/api",
+        "xz-5.2/src/liblzma/check",
+        "xz-5.2/src/liblzma/common",
+        "xz-5.2/src/liblzma/delta",
+        "xz-5.2/src/liblzma/lz",
+        "xz-5.2/src/liblzma/lzma",
+        "xz-5.2/src/liblzma/rangecoder",
+        "xz-5.2/src/liblzma/simple",
+        # The current working directory.
+        "",
+    ],
+    local_defines = ["HAVE_CONFIG_H=1"],
+)

src/adapter/src/active_compute_sink.rs

@@ -29,7 +29,7 @@ use tokio::sync::{mpsc, oneshot};
 use uuid::Uuid;
 
 use crate::coord::peek::PeekResponseUnary;
-use crate::{AdapterError, ExecuteResponse};
+use crate::{AdapterError, ExecuteContext, ExecuteResponse};
 
 #[derive(Debug)]
 /// A description of an active compute sink from the coordinator's perspective.
@@ -435,3 +435,12 @@ impl ActiveCopyTo {
         let _ = self.tx.send(message);
     }
 }
+
+/// State we keep in the `Coordinator` to track active `COPY FROM` statements.
+#[derive(Debug)]
+pub(crate) struct ActiveCopyFrom {
+    /// ID of the ingestion running in clusterd.
+    pub ingestion_id: uuid::Uuid,
+    /// Context of the SQL session that ran the statement.
+    pub ctx: ExecuteContext,
+}

src/adapter/src/coord.rs

@@ -170,7 +170,7 @@ use tracing::{debug, info, info_span, span, warn, Instrument, Level, Span};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 use uuid::Uuid;
 
-use crate::active_compute_sink::ActiveComputeSink;
+use crate::active_compute_sink::{ActiveComputeSink, ActiveCopyFrom};
 use crate::catalog::{BuiltinTableUpdate, Catalog, OpenCatalogResult};
 use crate::client::{Client, Handle};
 use crate::command::{Command, ExecuteResponse};
@@ -1667,7 +1667,7 @@ pub struct Coordinator {
     active_webhooks: BTreeMap<CatalogItemId, WebhookAppenderInvalidator>,
     /// A map of active `COPY FROM` statements. The Coordinator waits for `clusterd`
     /// to stage Batches in Persist that we will then link into the shard.
-    active_copies: BTreeMap<ConnectionId, ExecuteContext>,
+    active_copies: BTreeMap<ConnectionId, ActiveCopyFrom>,
 
     /// A map from connection ids to a watch channel that is set to `true` if the connection
     /// received a cancel request.

src/adapter/src/coord/sequencer.rs

@@ -373,7 +373,7 @@ impl Coordinator {
                             session,
                         );
                     }
-                    CopyFromSource::Url(_) => {
+                    CopyFromSource::Url(_) | CopyFromSource::AwsS3 { .. } => {
                         self.sequence_copy_from(ctx, plan, target_cluster).await;
                     }
                 },