(table: &Table, commit_user: String)
| 110 | |
| 111 | impl TableWrite { |
| 112 | pub(crate) fn new(table: &Table, commit_user: String) -> crate::Result<Self> { |
| 113 | let is_overwrite = false; |
| 114 | let schema = table.schema(); |
| 115 | let core_options = CoreOptions::new(schema.options()); |
| 116 | let blob_descriptor_fields = core_options.blob_descriptor_fields(); |
| 117 | |
| 118 | for name in &blob_descriptor_fields { |
| 119 | match schema.fields().iter().find(|f| f.name() == name) { |
| 120 | None => { |
| 121 | return Err(crate::Error::DataInvalid { |
| 122 | message: format!("blob-descriptor-field '{name}' does not exist in schema"), |
| 123 | source: None, |
| 124 | }); |
| 125 | } |
| 126 | Some(f) if !f.data_type().is_blob_type() => { |
| 127 | return Err(crate::Error::DataInvalid { |
| 128 | message: format!( |
| 129 | "blob-descriptor-field '{name}' is not a top-level BLOB field" |
| 130 | ), |
| 131 | source: None, |
| 132 | }); |
| 133 | } |
| 134 | _ => {} |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | let total_buckets = core_options.bucket(); |
| 139 | let has_primary_keys = !schema.primary_keys().is_empty(); |
| 140 | let is_dynamic_bucket = has_primary_keys && total_buckets == -1; |
| 141 | |
| 142 | let is_dynamic_cross_partition = |
| 143 | is_dynamic_bucket && !schema.partition_keys().is_empty() && { |
| 144 | let pk_set: HashSet<&str> = |
| 145 | schema.primary_keys().iter().map(String::as_str).collect(); |
| 146 | schema |
| 147 | .partition_keys() |
| 148 | .iter() |
| 149 | .any(|p| !pk_set.contains(p.as_str())) |
| 150 | }; |
| 151 | |
| 152 | if has_primary_keys |
| 153 | && !is_dynamic_bucket |
| 154 | && total_buckets < 1 |
| 155 | && total_buckets != POSTPONE_BUCKET |
| 156 | { |
| 157 | return Err(crate::Error::Unsupported { |
| 158 | message: format!( |
| 159 | "KeyValueFileWriter does not support bucket={total_buckets}, only fixed bucket (>= 1), -1 (dynamic), or -2 (postpone) is supported" |
| 160 | ), |
| 161 | }); |
| 162 | } |
| 163 | if has_primary_keys |
| 164 | && total_buckets != POSTPONE_BUCKET |
| 165 | && core_options |
| 166 | .changelog_producer() |
| 167 | .eq_ignore_ascii_case("input") |
| 168 | { |
| 169 | return Err(crate::Error::Unsupported { |
nothing calls this directly
no test coverage detected