Line data Source code
1 : /******************************************************************************
2 : *
3 : * Project: Parquet Translator
4 : * Purpose: Implements OGRParquetDriver.
5 : * Author: Even Rouault, <even.rouault at spatialys.com>
6 : *
7 : ******************************************************************************
8 : * Copyright (c) 2022, Planet Labs
9 : *
10 : * SPDX-License-Identifier: MIT
11 : ****************************************************************************/
12 :
13 : #ifdef STANDALONE
14 : #include "gdal_version.h"
15 : #else
16 : #undef DO_NOT_DEFINE_GDAL_DATE_NAME
17 : #include "gdal_version_full/gdal_version.h"
18 : #endif
19 :
20 : #include "ogr_parquet.h"
21 :
22 : #include "../arrow_common/ograrrowwriterlayer.hpp"
23 :
24 : #include "ogr_wkb.h"
25 :
26 : #include <cassert>
27 : #include <utility>
28 :
29 : /************************************************************************/
30 : /* OGRParquetWriterLayer() */
31 : /************************************************************************/
32 :
33 364 : OGRParquetWriterLayer::OGRParquetWriterLayer(
34 : OGRParquetWriterDataset *poDataset, arrow::MemoryPool *poMemoryPool,
35 : const std::shared_ptr<arrow::io::OutputStream> &poOutputStream,
36 364 : const char *pszLayerName)
37 : : OGRArrowWriterLayer(poMemoryPool, poOutputStream, pszLayerName),
38 364 : m_poDataset(poDataset)
39 : {
40 364 : m_bWriteFieldArrowExtensionName = CPLTestBool(
41 : CPLGetConfigOption("OGR_PARQUET_WRITE_ARROW_EXTENSION_NAME", "NO"));
42 364 : }
43 :
44 : /************************************************************************/
45 : /* Close() */
46 : /************************************************************************/
47 :
48 361 : bool OGRParquetWriterLayer::Close()
49 : {
50 361 : if (m_poTmpGPKGLayer)
51 : {
52 3 : if (!CopyTmpGpkgLayerToFinalFile())
53 0 : return false;
54 : }
55 :
56 361 : if (m_bInitializationOK)
57 : {
58 361 : if (!FinalizeWriting())
59 0 : return false;
60 : }
61 :
62 361 : return true;
63 : }
64 :
65 : /************************************************************************/
66 : /* CopyTmpGpkgLayerToFinalFile() */
67 : /************************************************************************/
68 :
69 3 : bool OGRParquetWriterLayer::CopyTmpGpkgLayerToFinalFile()
70 : {
71 3 : if (!m_poTmpGPKGLayer)
72 : {
73 0 : return true;
74 : }
75 :
76 3 : CPLDebug("PARQUET", "CopyTmpGpkgLayerToFinalFile(): start...");
77 :
78 3 : VSIUnlink(m_poTmpGPKG->GetDescription());
79 :
80 6 : OGRFeature oFeat(m_poFeatureDefn);
81 :
82 : // Interval in terms of features between 2 debug progress report messages
83 3 : constexpr int PROGRESS_FC_INTERVAL = 100 * 1000;
84 :
85 : // First, write features without geometries
86 : {
87 3 : auto poTmpLayer = std::unique_ptr<OGRLayer>(m_poTmpGPKG->ExecuteSQL(
88 : "SELECT serialized_feature FROM tmp WHERE fid NOT IN (SELECT id "
89 : "FROM rtree_tmp_geom)",
90 3 : nullptr, nullptr));
91 3 : if (!poTmpLayer)
92 0 : return false;
93 1005 : for (const auto &poSrcFeature : poTmpLayer.get())
94 : {
95 1002 : int nBytesFeature = 0;
96 : const GByte *pabyFeatureData =
97 1002 : poSrcFeature->GetFieldAsBinary(0, &nBytesFeature);
98 1002 : if (!oFeat.DeserializeFromBinary(pabyFeatureData, nBytesFeature))
99 : {
100 0 : CPLError(CE_Failure, CPLE_AppDefined,
101 : "Cannot deserialize feature");
102 0 : return false;
103 : }
104 1002 : if (OGRArrowWriterLayer::ICreateFeature(&oFeat) != OGRERR_NONE)
105 : {
106 0 : return false;
107 : }
108 :
109 1002 : if ((m_nFeatureCount % PROGRESS_FC_INTERVAL) == 0)
110 : {
111 0 : CPLDebugProgress(
112 : "PARQUET",
113 : "CopyTmpGpkgLayerToFinalFile(): %.02f%% progress",
114 0 : 100.0 * double(m_nFeatureCount) /
115 0 : double(m_nTmpFeatureCount));
116 : }
117 : }
118 :
119 3 : if (!FlushFeatures())
120 : {
121 0 : return false;
122 : }
123 : }
124 :
125 : // Now walk through the GPKG RTree for features with geometries
126 : // Cf https://github.com/sqlite/sqlite/blob/master/ext/rtree/rtree.c
127 : // for the description of the content of the rtree _node table
128 6 : std::vector<std::pair<int64_t, int>> aNodeNoDepthPair;
129 3 : int nTreeDepth = 0;
130 : // Queue the root node
131 : aNodeNoDepthPair.emplace_back(
132 3 : std::make_pair(/* nodeNo = */ 1, /* depth = */ 0));
133 3 : int nCountWrittenFeaturesSinceLastFlush = 0;
134 50 : while (!aNodeNoDepthPair.empty())
135 : {
136 47 : const auto &oLastPair = aNodeNoDepthPair.back();
137 47 : const int64_t nNodeNo = oLastPair.first;
138 47 : const int nCurDepth = oLastPair.second;
139 : //CPLDebug("PARQUET", "Reading nodeNode=%d, curDepth=%d", int(nNodeNo), nCurDepth);
140 47 : aNodeNoDepthPair.pop_back();
141 :
142 47 : auto poRTreeLayer = std::unique_ptr<OGRLayer>(m_poTmpGPKG->ExecuteSQL(
143 : CPLSPrintf("SELECT data FROM rtree_tmp_geom_node WHERE nodeno "
144 : "= " CPL_FRMT_GIB,
145 : static_cast<GIntBig>(nNodeNo)),
146 47 : nullptr, nullptr));
147 47 : if (!poRTreeLayer)
148 : {
149 0 : CPLError(CE_Failure, CPLE_AppDefined,
150 : "Cannot read node " CPL_FRMT_GIB,
151 : static_cast<GIntBig>(nNodeNo));
152 0 : return false;
153 : }
154 : const auto poRTreeFeature =
155 47 : std::unique_ptr<const OGRFeature>(poRTreeLayer->GetNextFeature());
156 47 : if (!poRTreeFeature)
157 : {
158 0 : CPLError(CE_Failure, CPLE_AppDefined,
159 : "Cannot read node " CPL_FRMT_GIB,
160 : static_cast<GIntBig>(nNodeNo));
161 0 : return false;
162 : }
163 :
164 47 : int nNodeBytes = 0;
165 : const GByte *pabyNodeData =
166 47 : poRTreeFeature->GetFieldAsBinary(0, &nNodeBytes);
167 47 : constexpr int BLOB_HEADER_SIZE = 4;
168 47 : if (nNodeBytes < BLOB_HEADER_SIZE)
169 : {
170 0 : CPLError(CE_Failure, CPLE_AppDefined,
171 : "Not enough bytes when reading node " CPL_FRMT_GIB,
172 : static_cast<GIntBig>(nNodeNo));
173 0 : return false;
174 : }
175 47 : if (nNodeNo == 1)
176 : {
177 : // Get the RTree depth from the root node
178 3 : nTreeDepth = (pabyNodeData[0] << 8) | pabyNodeData[1];
179 : //CPLDebug("PARQUET", "nTreeDepth = %d", nTreeDepth);
180 : }
181 :
182 47 : const int nCellCount = (pabyNodeData[2] << 8) | pabyNodeData[3];
183 47 : constexpr int SIZEOF_CELL = 24; // int64_t + 4 float
184 47 : if (nNodeBytes < BLOB_HEADER_SIZE + SIZEOF_CELL * nCellCount)
185 : {
186 0 : CPLError(CE_Failure, CPLE_AppDefined,
187 : "Not enough bytes when reading node " CPL_FRMT_GIB,
188 : static_cast<GIntBig>(nNodeNo));
189 0 : return false;
190 : }
191 :
192 47 : size_t nOffset = BLOB_HEADER_SIZE;
193 47 : if (nCurDepth == nTreeDepth)
194 : {
195 : // Leaf node: it references feature IDs.
196 :
197 : // If we are about to go above m_nRowGroupSize, flush past
198 : // features now, to improve the spatial compacity of the row group.
199 45 : if (m_nRowGroupSize > nCellCount &&
200 45 : nCountWrittenFeaturesSinceLastFlush + nCellCount >
201 45 : m_nRowGroupSize)
202 : {
203 14 : nCountWrittenFeaturesSinceLastFlush = 0;
204 14 : if (!FlushFeatures())
205 : {
206 0 : return false;
207 : }
208 : }
209 :
210 : // nCellCount shouldn't be over 51 normally, but even 65535
211 : // would be fine...
212 45 : assert(nCellCount <= 65535);
213 1247 : for (int i = 0; i < nCellCount; ++i)
214 : {
215 : int64_t nFID;
216 1202 : memcpy(&nFID, pabyNodeData + nOffset, sizeof(int64_t));
217 1202 : CPL_MSBPTR64(&nFID);
218 :
219 : const auto poSrcFeature = std::unique_ptr<const OGRFeature>(
220 1202 : m_poTmpGPKGLayer->GetFeature(nFID));
221 1202 : if (!poSrcFeature)
222 : {
223 0 : CPLError(CE_Failure, CPLE_AppDefined,
224 : "Cannot get feature " CPL_FRMT_GIB,
225 : static_cast<GIntBig>(nFID));
226 0 : return false;
227 : }
228 :
229 1202 : int nBytesFeature = 0;
230 : const GByte *pabyFeatureData =
231 1202 : poSrcFeature->GetFieldAsBinary(0, &nBytesFeature);
232 1202 : if (!oFeat.DeserializeFromBinary(pabyFeatureData,
233 : nBytesFeature))
234 : {
235 0 : CPLError(CE_Failure, CPLE_AppDefined,
236 : "Cannot deserialize feature");
237 0 : return false;
238 : }
239 1202 : if (OGRArrowWriterLayer::ICreateFeature(&oFeat) != OGRERR_NONE)
240 : {
241 0 : return false;
242 : }
243 :
244 1202 : nOffset += SIZEOF_CELL;
245 :
246 1202 : ++nCountWrittenFeaturesSinceLastFlush;
247 :
248 1202 : if ((m_nFeatureCount % PROGRESS_FC_INTERVAL) == 0 ||
249 1202 : m_nFeatureCount == m_nTmpFeatureCount / 2)
250 : {
251 2 : CPLDebugProgress(
252 : "PARQUET",
253 : "CopyTmpGpkgLayerToFinalFile(): %.02f%% progress",
254 2 : 100.0 * double(m_nFeatureCount) /
255 2 : double(m_nTmpFeatureCount));
256 : }
257 : }
258 : }
259 : else
260 : {
261 : // Non-leaf node: it references child nodes.
262 :
263 : // nCellCount shouldn't be over 51 normally, but even 65535
264 : // would be fine...
265 2 : assert(nCellCount <= 65535);
266 46 : for (int i = 0; i < nCellCount; ++i)
267 : {
268 : int64_t nNode;
269 44 : memcpy(&nNode, pabyNodeData + nOffset, sizeof(int64_t));
270 44 : CPL_MSBPTR64(&nNode);
271 : aNodeNoDepthPair.emplace_back(
272 44 : std::make_pair(nNode, nCurDepth + 1));
273 44 : nOffset += SIZEOF_CELL;
274 : }
275 : }
276 : }
277 :
278 3 : CPLDebug("PARQUET",
279 : "CopyTmpGpkgLayerToFinalFile(): 100%%, successfully finished");
280 3 : return true;
281 : }
282 :
283 : /************************************************************************/
284 : /* IsSupportedGeometryType() */
285 : /************************************************************************/
286 :
287 360 : bool OGRParquetWriterLayer::IsSupportedGeometryType(
288 : OGRwkbGeometryType eGType) const
289 : {
290 360 : const auto eFlattenType = wkbFlatten(eGType);
291 360 : if (!OGR_GT_HasM(eGType) && eFlattenType <= wkbGeometryCollection)
292 : {
293 359 : return true;
294 : }
295 :
296 : const auto osConfigOptionName =
297 3 : "OGR_" + GetDriverUCName() + "_ALLOW_ALL_DIMS";
298 1 : if (CPLTestBool(CPLGetConfigOption(osConfigOptionName.c_str(), "NO")))
299 : {
300 0 : return true;
301 : }
302 :
303 1 : CPLError(CE_Failure, CPLE_NotSupported,
304 : "Only 2D and Z geometry types are supported (unless the "
305 : "%s configuration option is set to YES)",
306 : osConfigOptionName.c_str());
307 1 : return false;
308 : }
309 :
310 : /************************************************************************/
311 : /* SetOptions() */
312 : /************************************************************************/
313 :
314 364 : bool OGRParquetWriterLayer::SetOptions(
315 : const OGRGeomFieldDefn *poSrcGeomFieldDefn, CSLConstList papszOptions)
316 : {
317 364 : m_aosCreationOptions = papszOptions;
318 :
319 364 : const char *pszWriteCoveringBBox = CSLFetchNameValueDef(
320 : papszOptions, "WRITE_COVERING_BBOX",
321 : CPLGetConfigOption("OGR_PARQUET_WRITE_COVERING_BBOX", nullptr));
322 364 : m_bWriteBBoxStruct =
323 364 : pszWriteCoveringBBox == nullptr || CPLTestBool(pszWriteCoveringBBox);
324 :
325 : m_oBBoxStructFieldName =
326 364 : CSLFetchNameValueDef(papszOptions, "COVERING_BBOX_NAME", "");
327 :
328 364 : if (CPLTestBool(CSLFetchNameValueDef(papszOptions, "SORT_BY_BBOX", "NO")))
329 : {
330 8 : const std::string osTmpGPKG(std::string(m_poDataset->GetDescription()) +
331 4 : ".tmp.gpkg");
332 4 : auto poGPKGDrv = GetGDALDriverManager()->GetDriverByName("GPKG");
333 4 : if (!poGPKGDrv)
334 : {
335 1 : CPLError(
336 : CE_Failure, CPLE_AppDefined,
337 : "Driver GPKG required for SORT_BY_BBOX layer creation option");
338 1 : return false;
339 : }
340 3 : m_poTmpGPKG.reset(poGPKGDrv->Create(osTmpGPKG.c_str(), 0, 0, 0,
341 : GDT_Unknown, nullptr));
342 3 : if (!m_poTmpGPKG)
343 0 : return false;
344 3 : m_poTmpGPKG->MarkSuppressOnClose();
345 3 : m_poTmpGPKGLayer = m_poTmpGPKG->CreateLayer("tmp");
346 12 : if (!m_poTmpGPKGLayer ||
347 : // Serialized feature
348 3 : m_poTmpGPKGLayer->CreateField(
349 3 : std::make_unique<OGRFieldDefn>("serialized_feature", OFTBinary)
350 3 : .get()) != OGRERR_NONE ||
351 : // FlushCache is needed to avoid SQLite3 errors on empty layers
352 9 : m_poTmpGPKG->FlushCache() != CE_None ||
353 3 : m_poTmpGPKGLayer->StartTransaction() != OGRERR_NONE)
354 : {
355 0 : return false;
356 : }
357 : }
358 :
359 : const char *pszGeomEncoding =
360 363 : CSLFetchNameValue(papszOptions, "GEOMETRY_ENCODING");
361 363 : m_eGeomEncoding = OGRArrowGeomEncoding::WKB;
362 363 : if (pszGeomEncoding)
363 : {
364 211 : if (EQUAL(pszGeomEncoding, "WKB"))
365 7 : m_eGeomEncoding = OGRArrowGeomEncoding::WKB;
366 204 : else if (EQUAL(pszGeomEncoding, "WKT"))
367 8 : m_eGeomEncoding = OGRArrowGeomEncoding::WKT;
368 196 : else if (EQUAL(pszGeomEncoding, "GEOARROW_INTERLEAVED"))
369 : {
370 28 : CPLErrorOnce(
371 : CE_Warning, CPLE_AppDefined,
372 : "Use of GEOMETRY_ENCODING=GEOARROW_INTERLEAVED is not "
373 : "recommended. "
374 : "GeoParquet 1.1 uses GEOMETRY_ENCODING=GEOARROW (struct) "
375 : "instead.");
376 28 : m_eGeomEncoding = OGRArrowGeomEncoding::GEOARROW_FSL_GENERIC;
377 : }
378 168 : else if (EQUAL(pszGeomEncoding, "GEOARROW") ||
379 0 : EQUAL(pszGeomEncoding, "GEOARROW_STRUCT"))
380 168 : m_eGeomEncoding = OGRArrowGeomEncoding::GEOARROW_STRUCT_GENERIC;
381 : else
382 : {
383 0 : CPLError(CE_Failure, CPLE_NotSupported,
384 : "Unsupported GEOMETRY_ENCODING = %s", pszGeomEncoding);
385 0 : return false;
386 : }
387 : }
388 :
389 : const char *pszCoordPrecision =
390 363 : CSLFetchNameValue(papszOptions, "COORDINATE_PRECISION");
391 363 : if (pszCoordPrecision)
392 0 : m_nWKTCoordinatePrecision = atoi(pszCoordPrecision);
393 :
394 363 : m_bForceCounterClockwiseOrientation =
395 363 : EQUAL(CSLFetchNameValueDef(papszOptions, "POLYGON_ORIENTATION",
396 : "COUNTERCLOCKWISE"),
397 : "COUNTERCLOCKWISE");
398 :
399 : const auto eGType =
400 363 : poSrcGeomFieldDefn ? poSrcGeomFieldDefn->GetType() : wkbNone;
401 363 : if (poSrcGeomFieldDefn && eGType != wkbNone)
402 : {
403 334 : if (!IsSupportedGeometryType(eGType))
404 : {
405 1 : return false;
406 : }
407 :
408 333 : m_poFeatureDefn->SetGeomType(eGType);
409 333 : auto eGeomEncoding = m_eGeomEncoding;
410 333 : if (eGeomEncoding == OGRArrowGeomEncoding::GEOARROW_FSL_GENERIC ||
411 305 : eGeomEncoding == OGRArrowGeomEncoding::GEOARROW_STRUCT_GENERIC)
412 : {
413 196 : const auto eEncodingType = eGeomEncoding;
414 196 : eGeomEncoding = GetPreciseArrowGeomEncoding(eEncodingType, eGType);
415 196 : if (eGeomEncoding == eEncodingType)
416 0 : return false;
417 : }
418 333 : m_aeGeomEncoding.push_back(eGeomEncoding);
419 :
420 666 : std::string osGeometryName;
421 : const char *pszGeometryName =
422 333 : CSLFetchNameValue(papszOptions, "GEOMETRY_NAME");
423 333 : if (pszGeometryName)
424 16 : osGeometryName = pszGeometryName;
425 317 : else if (poSrcGeomFieldDefn->GetNameRef()[0])
426 4 : osGeometryName = poSrcGeomFieldDefn->GetNameRef();
427 : else
428 313 : osGeometryName = "geometry";
429 333 : m_poFeatureDefn->GetGeomFieldDefn(0)->SetName(osGeometryName.c_str());
430 :
431 333 : const auto poSpatialRef = poSrcGeomFieldDefn->GetSpatialRef();
432 333 : if (poSpatialRef)
433 : {
434 42 : auto poSRS = poSpatialRef->Clone();
435 42 : m_poFeatureDefn->GetGeomFieldDefn(0)->SetSpatialRef(poSRS);
436 42 : poSRS->Release();
437 : }
438 : }
439 :
440 362 : m_osFIDColumn = CSLFetchNameValueDef(papszOptions, "FID", "");
441 :
442 : const char *pszCompression =
443 362 : CSLFetchNameValue(papszOptions, GDALMD_COMPRESSION);
444 362 : if (pszCompression == nullptr)
445 : {
446 1068 : auto oResult = arrow::util::Codec::GetCompressionType("snappy");
447 356 : if (oResult.ok() && arrow::util::Codec::IsAvailable(*oResult))
448 : {
449 356 : pszCompression = "SNAPPY";
450 : }
451 : else
452 : {
453 0 : pszCompression = "NONE";
454 : }
455 : }
456 :
457 362 : if (EQUAL(pszCompression, "NONE"))
458 0 : pszCompression = "UNCOMPRESSED";
459 : auto oResult = arrow::util::Codec::GetCompressionType(
460 724 : CPLString(pszCompression).tolower());
461 362 : if (!oResult.ok())
462 : {
463 1 : CPLError(CE_Failure, CPLE_NotSupported,
464 : "Unrecognized compression method: %s", pszCompression);
465 1 : return false;
466 : }
467 361 : m_eCompression = *oResult;
468 361 : if (!arrow::util::Codec::IsAvailable(m_eCompression))
469 : {
470 0 : CPLError(CE_Failure, CPLE_NotSupported,
471 : "Compression method %s is known, but libarrow has not "
472 : "been built with support for it",
473 : pszCompression);
474 0 : return false;
475 : }
476 361 : m_oWriterPropertiesBuilder.compression(m_eCompression);
477 :
478 : const char *pszCompressionLevel =
479 361 : CSLFetchNameValue(papszOptions, "COMPRESSION_LEVEL");
480 361 : if (pszCompressionLevel)
481 : {
482 2 : const int nCompressionLevel = atoi(pszCompressionLevel);
483 2 : if (nCompressionLevel != DEFAULT_COMPRESSION_LEVEL)
484 2 : m_oWriterPropertiesBuilder.compression_level(nCompressionLevel);
485 : }
486 359 : else if (EQUAL(pszCompression, "ZSTD"))
487 1 : m_oWriterPropertiesBuilder.compression_level(
488 : OGR_PARQUET_ZSTD_DEFAULT_COMPRESSION_LEVEL);
489 :
490 : const std::string osCreator =
491 722 : CSLFetchNameValueDef(papszOptions, "CREATOR", "");
492 361 : if (!osCreator.empty())
493 1 : m_oWriterPropertiesBuilder.created_by(osCreator);
494 : else
495 360 : m_oWriterPropertiesBuilder.created_by("GDAL " GDAL_RELEASE_NAME
496 : ", using " CREATED_BY_VERSION);
497 :
498 : // Undocumented option. Not clear it is useful besides unit test purposes
499 361 : if (!CPLTestBool(CSLFetchNameValueDef(papszOptions, "STATISTICS", "YES")))
500 1 : m_oWriterPropertiesBuilder.disable_statistics();
501 :
502 : #if PARQUET_VERSION_MAJOR >= 12
503 : // Undocumented option. Not clear it is useful to disable it.
504 361 : if (CPLTestBool(CSLFetchNameValueDef(papszOptions, "PAGE_INDEX", "YES")))
505 361 : m_oWriterPropertiesBuilder.enable_write_page_index();
506 : #endif
507 :
508 : const char *pszGeoParquetVersion =
509 361 : CSLFetchNameValueDef(papszOptions, "GEOPARQUET_VERSION", "1.1");
510 361 : if (EQUAL(pszGeoParquetVersion, "1.1") ||
511 0 : EQUAL(pszGeoParquetVersion, "AUTO"))
512 361 : m_nGeoParquetVersion = OGRGeoParquetVersion::VERSION_1_1;
513 0 : else if (EQUAL(pszGeoParquetVersion, "2.0"))
514 : {
515 : #if ARROW_VERSION_MAJOR >= 21
516 : m_nGeoParquetVersion = OGRGeoParquetVersion::VERSION_2_0;
517 : if (pszWriteCoveringBBox == nullptr)
518 : m_bWriteBBoxStruct = false;
519 : #else
520 0 : CPLError(CE_Failure, CPLE_NotSupported,
521 : "GEOPARQUET_VERSION = 2.0 is only supported in a GDAL build "
522 : "against libarrow >= 21");
523 0 : return false;
524 : #endif
525 : }
526 : else
527 : {
528 0 : CPLError(CE_Failure, CPLE_NotSupported,
529 : "Unrecognized GeoParquet version: %s", pszGeoParquetVersion);
530 0 : return false;
531 : }
532 :
533 : const char *pszWriteGeo =
534 361 : CPLGetConfigOption("OGR_PARQUET_WRITE_GEO", nullptr);
535 361 : m_bWriteGeoMetadata = pszWriteGeo == nullptr || CPLTestBool(pszWriteGeo);
536 :
537 361 : if (m_eGeomEncoding == OGRArrowGeomEncoding::WKB && eGType != wkbNone)
538 : {
539 : #if ARROW_VERSION_MAJOR >= 21
540 : const char *pszUseParquetGeoTypes =
541 : CSLFetchNameValueDef(papszOptions, "USE_PARQUET_GEO_TYPES", "AUTO");
542 : if (EQUAL(pszUseParquetGeoTypes, "AUTO"))
543 : {
544 : m_bUseArrowWKBExtension =
545 : (m_nGeoParquetVersion == OGRGeoParquetVersion::VERSION_2_0);
546 : }
547 : else if (EQUAL(pszUseParquetGeoTypes, "ONLY"))
548 : {
549 : m_bUseArrowWKBExtension = true;
550 : if (pszWriteGeo == nullptr)
551 : m_bWriteGeoMetadata = false;
552 : if (pszWriteCoveringBBox == nullptr)
553 : m_bWriteBBoxStruct = false;
554 : }
555 : else
556 : {
557 : m_bUseArrowWKBExtension = CPLTestBool(pszUseParquetGeoTypes);
558 : if (!m_bUseArrowWKBExtension &&
559 : m_nGeoParquetVersion == OGRGeoParquetVersion::VERSION_2_0)
560 : {
561 : CPLError(CE_Failure, CPLE_NotSupported,
562 : "GEOPARQUET_VERSION = 2.0 is not compatible with "
563 : "USE_PARQUET_GEO_TYPES = NO");
564 : return false;
565 : }
566 : }
567 : #else
568 136 : m_oWriterPropertiesBuilder.disable_statistics(
569 408 : parquet::schema::ColumnPath::FromDotString(
570 136 : m_poFeatureDefn->GetGeomFieldDefn(0)->GetNameRef()));
571 : #endif
572 : }
573 :
574 : const char *pszRowGroupSize =
575 361 : CSLFetchNameValue(papszOptions, "ROW_GROUP_SIZE");
576 361 : if (pszRowGroupSize)
577 : {
578 19 : auto nRowGroupSize = static_cast<int64_t>(atoll(pszRowGroupSize));
579 19 : if (nRowGroupSize > 0)
580 : {
581 19 : if (nRowGroupSize > INT_MAX)
582 0 : nRowGroupSize = INT_MAX;
583 19 : m_nRowGroupSize = nRowGroupSize;
584 : }
585 : }
586 :
587 361 : m_bEdgesSpherical = EQUAL(
588 : CSLFetchNameValueDef(papszOptions, "EDGES", "PLANAR"), "SPHERICAL");
589 :
590 361 : m_bInitializationOK = true;
591 361 : return true;
592 : }
593 :
594 : /************************************************************************/
595 : /* CloseFileWriter() */
596 : /************************************************************************/
597 :
598 361 : bool OGRParquetWriterLayer::CloseFileWriter()
599 : {
600 722 : auto status = m_poFileWriter->Close();
601 361 : if (!status.ok())
602 : {
603 0 : CPLError(CE_Failure, CPLE_AppDefined,
604 : "FileWriter::Close() failed with %s",
605 0 : status.message().c_str());
606 : }
607 722 : return status.ok();
608 : }
609 :
610 : /************************************************************************/
611 : /* GetGeoMetadata() */
612 : /************************************************************************/
613 :
614 361 : std::string OGRParquetWriterLayer::GetGeoMetadata() const
615 : {
616 : // Just for unit testing purposes
617 : const char *pszGeoMetadata =
618 361 : CPLGetConfigOption("OGR_PARQUET_GEO_METADATA", nullptr);
619 361 : if (pszGeoMetadata)
620 16 : return pszGeoMetadata;
621 :
622 345 : if (m_poFeatureDefn->GetGeomFieldCount() != 0 && m_bWriteGeoMetadata)
623 : {
624 650 : CPLJSONObject oRoot;
625 325 : oRoot.Add("version",
626 325 : m_nGeoParquetVersion == OGRGeoParquetVersion::VERSION_1_1
627 : ? "1.1.0"
628 : : "2.0.0");
629 325 : oRoot.Add("primary_column",
630 325 : m_poFeatureDefn->GetGeomFieldDefn(0)->GetNameRef());
631 650 : CPLJSONObject oColumns;
632 325 : oRoot.Add("columns", oColumns);
633 667 : for (int i = 0; i < m_poFeatureDefn->GetGeomFieldCount(); ++i)
634 : {
635 342 : const auto poGeomFieldDefn = m_poFeatureDefn->GetGeomFieldDefn(i);
636 684 : CPLJSONObject oColumn;
637 342 : oColumns.Add(poGeomFieldDefn->GetNameRef(), oColumn);
638 342 : oColumn.Add("encoding",
639 342 : GetGeomEncodingAsString(m_aeGeomEncoding[i], true));
640 :
641 342 : if (CPLTestBool(CPLGetConfigOption("OGR_PARQUET_WRITE_CRS", "YES")))
642 : {
643 341 : const auto poSRS = poGeomFieldDefn->GetSpatialRef();
644 341 : if (poSRS)
645 : {
646 82 : OGRSpatialReference oSRSIdentified(IdentifyCRS(poSRS));
647 :
648 41 : const char *pszAuthName = oSRSIdentified.GetAuthorityName();
649 41 : const char *pszAuthCode = oSRSIdentified.GetAuthorityCode();
650 :
651 41 : bool bOmitCRS = false;
652 41 : if (pszAuthName != nullptr && pszAuthCode != nullptr &&
653 40 : ((EQUAL(pszAuthName, "EPSG") &&
654 37 : EQUAL(pszAuthCode, "4326")) ||
655 22 : (EQUAL(pszAuthName, "OGC") &&
656 3 : EQUAL(pszAuthCode, "CRS84"))))
657 : {
658 : // To make things less confusing for non-geo-aware
659 : // consumers, omit EPSG:4326 / OGC:CRS84 CRS by default
660 21 : bOmitCRS = CPLTestBool(CPLGetConfigOption(
661 : "OGR_PARQUET_CRS_OMIT_IF_WGS84", "YES"));
662 : }
663 :
664 41 : if (bOmitCRS)
665 : {
666 : // do nothing
667 : }
668 20 : else if (EQUAL(CPLGetConfigOption(
669 : "OGR_PARQUET_CRS_ENCODING", "PROJJSON"),
670 : "PROJJSON"))
671 : {
672 : // CRS encoded as PROJJSON for GeoParquet >= 0.4.0
673 20 : char *pszPROJJSON = nullptr;
674 20 : oSRSIdentified.exportToPROJJSON(&pszPROJJSON, nullptr);
675 40 : CPLJSONDocument oCRSDoc;
676 20 : CPL_IGNORE_RET_VAL(oCRSDoc.LoadMemory(pszPROJJSON));
677 20 : CPLFree(pszPROJJSON);
678 20 : CPLJSONObject oCRSRoot = oCRSDoc.GetRoot();
679 20 : RemoveIDFromMemberOfEnsembles(oCRSRoot);
680 20 : oColumn.Add("crs", oCRSRoot);
681 : }
682 : else
683 : {
684 : // WKT was used in GeoParquet <= 0.3.0
685 0 : const char *const apszOptions[] = {
686 : "FORMAT=WKT2_2019", "MULTILINE=NO", nullptr};
687 0 : char *pszWKT = nullptr;
688 0 : oSRSIdentified.exportToWkt(&pszWKT, apszOptions);
689 0 : if (pszWKT)
690 0 : oColumn.Add("crs", pszWKT);
691 0 : CPLFree(pszWKT);
692 : }
693 :
694 41 : const double dfCoordEpoch = poSRS->GetCoordinateEpoch();
695 41 : if (dfCoordEpoch > 0)
696 2 : oColumn.Add("epoch", dfCoordEpoch);
697 : }
698 : else
699 : {
700 300 : oColumn.AddNull("crs");
701 : }
702 : }
703 :
704 342 : if (m_bEdgesSpherical)
705 : {
706 3 : oColumn.Add("edges", "spherical");
707 : }
708 :
709 643 : if (m_aoEnvelopes[i].IsInit() &&
710 301 : CPLTestBool(
711 : CPLGetConfigOption("OGR_PARQUET_WRITE_BBOX", "YES")))
712 : {
713 301 : bool bHasZ = false;
714 537 : for (const auto eGeomType : m_oSetWrittenGeometryTypes[i])
715 : {
716 343 : bHasZ = CPL_TO_BOOL(OGR_GT_HasZ(eGeomType));
717 343 : if (bHasZ)
718 107 : break;
719 : }
720 301 : CPLJSONArray oBBOX;
721 301 : oBBOX.Add(m_aoEnvelopes[i].MinX);
722 301 : oBBOX.Add(m_aoEnvelopes[i].MinY);
723 301 : if (bHasZ)
724 107 : oBBOX.Add(m_aoEnvelopes[i].MinZ);
725 301 : oBBOX.Add(m_aoEnvelopes[i].MaxX);
726 301 : oBBOX.Add(m_aoEnvelopes[i].MaxY);
727 301 : if (bHasZ)
728 107 : oBBOX.Add(m_aoEnvelopes[i].MaxZ);
729 301 : oColumn.Add("bbox", oBBOX);
730 : }
731 :
732 : // Bounding box column definition
733 614 : if (m_bWriteBBoxStruct &&
734 272 : CPLTestBool(CPLGetConfigOption(
735 : "OGR_PARQUET_WRITE_COVERING_BBOX_IN_METADATA", "YES")))
736 : {
737 544 : CPLJSONObject oCovering;
738 272 : oColumn.Add("covering", oCovering);
739 544 : CPLJSONObject oBBOX;
740 272 : oCovering.Add("bbox", oBBOX);
741 : const auto AddComponent =
742 3264 : [this, i, &oBBOX](const char *pszComponent)
743 : {
744 1088 : CPLJSONArray oArray;
745 1088 : oArray.Add(m_apoFieldsBBOX[i]->name());
746 1088 : oArray.Add(pszComponent);
747 1088 : oBBOX.Add(pszComponent, oArray);
748 1088 : };
749 272 : AddComponent("xmin");
750 272 : AddComponent("ymin");
751 272 : AddComponent("xmax");
752 272 : AddComponent("ymax");
753 : }
754 :
755 359 : const auto GetStringGeometryType = [](OGRwkbGeometryType eType)
756 : {
757 359 : const auto eFlattenType = wkbFlatten(eType);
758 359 : std::string osType = "Unknown";
759 359 : if (wkbPoint == eFlattenType)
760 85 : osType = "Point";
761 274 : else if (wkbLineString == eFlattenType)
762 42 : osType = "LineString";
763 232 : else if (wkbPolygon == eFlattenType)
764 70 : osType = "Polygon";
765 162 : else if (wkbMultiPoint == eFlattenType)
766 34 : osType = "MultiPoint";
767 128 : else if (wkbMultiLineString == eFlattenType)
768 37 : osType = "MultiLineString";
769 91 : else if (wkbMultiPolygon == eFlattenType)
770 86 : osType = "MultiPolygon";
771 5 : else if (wkbGeometryCollection == eFlattenType)
772 5 : osType = "GeometryCollection";
773 359 : if (osType != "Unknown")
774 : {
775 : // M and ZM not supported officially currently, but it
776 : // doesn't hurt to anticipate
777 359 : if (OGR_GT_HasZ(eType) && OGR_GT_HasM(eType))
778 8 : osType += " ZM";
779 351 : else if (OGR_GT_HasZ(eType))
780 115 : osType += " Z";
781 236 : else if (OGR_GT_HasM(eType))
782 8 : osType += " M";
783 : }
784 359 : return osType;
785 : };
786 :
787 342 : if (m_bForceCounterClockwiseOrientation)
788 341 : oColumn.Add("orientation", "counterclockwise");
789 :
790 342 : CPLJSONArray oArray;
791 701 : for (const auto eType : m_oSetWrittenGeometryTypes[i])
792 : {
793 359 : oArray.Add(GetStringGeometryType(eType));
794 : }
795 342 : oColumn.Add("geometry_types", oArray);
796 : }
797 :
798 325 : return oRoot.Format(CPLJSONObject::PrettyFormat::Plain);
799 : }
800 20 : return std::string();
801 : }
802 :
803 : /************************************************************************/
804 : /* PerformStepsBeforeFinalFlushGroup() */
805 : /************************************************************************/
806 :
807 361 : void OGRParquetWriterLayer::PerformStepsBeforeFinalFlushGroup()
808 : {
809 361 : if (m_poKeyValueMetadata)
810 : {
811 722 : std::string osGeoMetadata = GetGeoMetadata();
812 722 : auto poTmpSchema = m_poSchema;
813 361 : if (!osGeoMetadata.empty())
814 : {
815 : // HACK: it would be good for Arrow to provide a clean way to alter
816 : // key value metadata before finalizing.
817 : // We need to write metadata at end to write the bounding box.
818 341 : const_cast<arrow::KeyValueMetadata *>(m_poKeyValueMetadata.get())
819 341 : ->Append("geo", osGeoMetadata);
820 :
821 341 : auto kvMetadata = poTmpSchema->metadata()
822 15 : ? poTmpSchema->metadata()->Copy()
823 356 : : std::make_shared<arrow::KeyValueMetadata>();
824 341 : kvMetadata->Append("geo", std::move(osGeoMetadata));
825 341 : poTmpSchema = poTmpSchema->WithMetadata(kvMetadata);
826 : }
827 :
828 361 : if (CPLTestBool(
829 : CPLGetConfigOption("OGR_PARQUET_WRITE_ARROW_SCHEMA", "YES")))
830 : {
831 : auto status =
832 722 : ::arrow::ipc::SerializeSchema(*poTmpSchema, m_poMemoryPool);
833 361 : if (status.ok())
834 : {
835 : // The serialized schema is not UTF-8, which is required for
836 : // Thrift
837 722 : const std::string schema_as_string = (*status)->ToString();
838 : std::string schema_base64 =
839 361 : ::arrow::util::base64_encode(schema_as_string);
840 361 : static const std::string kArrowSchemaKey = "ARROW:schema";
841 : const_cast<arrow::KeyValueMetadata *>(
842 361 : m_poKeyValueMetadata.get())
843 361 : ->Append(kArrowSchemaKey, std::move(schema_base64));
844 : }
845 : }
846 :
847 : // Put GDAL metadata into a gdal:metadata domain
848 722 : CPLJSONObject oMultiMetadata;
849 361 : bool bHasMultiMetadata = false;
850 369 : auto &l_oMDMD = oMDMD.GetDomainList() && *(oMDMD.GetDomainList())
851 369 : ? oMDMD
852 353 : : m_poDataset->GetMultiDomainMetadata();
853 371 : for (CSLConstList papszDomainIter = l_oMDMD.GetDomainList();
854 371 : papszDomainIter && *papszDomainIter; ++papszDomainIter)
855 : {
856 10 : const char *pszDomain = *papszDomainIter;
857 10 : CSLConstList papszMD = l_oMDMD.GetMetadata(pszDomain);
858 10 : if (STARTS_WITH(pszDomain, "json:") && papszMD && papszMD[0])
859 : {
860 1 : CPLJSONDocument oDoc;
861 1 : if (oDoc.LoadMemory(papszMD[0]))
862 : {
863 1 : bHasMultiMetadata = true;
864 1 : oMultiMetadata.Add(pszDomain, oDoc.GetRoot());
865 1 : continue;
866 0 : }
867 : }
868 9 : else if (STARTS_WITH(pszDomain, "xml:") && papszMD && papszMD[0])
869 : {
870 1 : bHasMultiMetadata = true;
871 1 : oMultiMetadata.Add(pszDomain, papszMD[0]);
872 1 : continue;
873 : }
874 16 : CPLJSONObject oMetadata;
875 8 : bool bHasMetadata = false;
876 16 : for (CSLConstList papszMDIter = papszMD;
877 16 : papszMDIter && *papszMDIter; ++papszMDIter)
878 : {
879 8 : char *pszKey = nullptr;
880 8 : const char *pszValue = CPLParseNameValue(*papszMDIter, &pszKey);
881 8 : if (pszKey && pszValue)
882 : {
883 8 : bHasMetadata = true;
884 8 : bHasMultiMetadata = true;
885 8 : oMetadata.Add(pszKey, pszValue);
886 : }
887 8 : CPLFree(pszKey);
888 : }
889 8 : if (bHasMetadata)
890 8 : oMultiMetadata.Add(pszDomain, oMetadata);
891 : }
892 361 : if (bHasMultiMetadata)
893 : {
894 8 : const_cast<arrow::KeyValueMetadata *>(m_poKeyValueMetadata.get())
895 8 : ->Append(
896 : "gdal:metadata",
897 16 : oMultiMetadata.Format(CPLJSONObject::PrettyFormat::Plain));
898 : }
899 :
900 361 : if (!m_aosCreationOptions.empty())
901 : {
902 508 : CPLJSONObject oCreationOptions;
903 254 : bool bEmpty = true;
904 1120 : for (const auto &[key, value] :
905 1374 : cpl::IterateNameValue(m_aosCreationOptions))
906 : {
907 560 : if (!EQUAL(key, "FID") && !EQUAL(key, "GEOMETRY_NAME") &&
908 521 : !EQUAL(key, "EDGES"))
909 : {
910 517 : bEmpty = false;
911 517 : oCreationOptions.Add(key, value);
912 : }
913 : }
914 254 : if (!bEmpty)
915 : {
916 : const_cast<arrow::KeyValueMetadata *>(
917 240 : m_poKeyValueMetadata.get())
918 240 : ->Append("gdal:creation-options",
919 480 : oCreationOptions.Format(
920 : CPLJSONObject::PrettyFormat::Plain));
921 : }
922 : }
923 : }
924 361 : }
925 :
926 : /************************************************************************/
927 : /* Open() */
928 : /************************************************************************/
929 :
930 : // Same as parquet::arrow::FileWriter::Open(), except we also
931 : // return KeyValueMetadata
932 : static arrow::Status
933 361 : Open(const ::arrow::Schema &schema, ::arrow::MemoryPool *pool,
934 : std::shared_ptr<::arrow::io::OutputStream> sink,
935 : std::shared_ptr<parquet::WriterProperties> properties,
936 : std::shared_ptr<parquet::ArrowWriterProperties> arrow_properties,
937 : std::unique_ptr<parquet::arrow::FileWriter> *writer,
938 : std::shared_ptr<const arrow::KeyValueMetadata> *outMetadata)
939 : {
940 361 : std::shared_ptr<parquet::SchemaDescriptor> parquet_schema;
941 722 : RETURN_NOT_OK(parquet::arrow::ToParquetSchema(
942 : &schema, *properties, *arrow_properties, &parquet_schema));
943 :
944 : auto schema_node = std::static_pointer_cast<parquet::schema::GroupNode>(
945 722 : parquet_schema->schema_root());
946 :
947 361 : auto metadata = schema.metadata()
948 20 : ? schema.metadata()->Copy()
949 742 : : std::make_shared<arrow::KeyValueMetadata>();
950 361 : *outMetadata = metadata;
951 :
952 361 : std::unique_ptr<parquet::ParquetFileWriter> base_writer;
953 361 : PARQUET_CATCH_NOT_OK(base_writer = parquet::ParquetFileWriter::Open(
954 : std::move(sink), std::move(schema_node),
955 : std::move(properties), metadata));
956 :
957 361 : auto schema_ptr = std::make_shared<::arrow::Schema>(schema);
958 : return parquet::arrow::FileWriter::Make(
959 722 : pool, std::move(base_writer), std::move(schema_ptr),
960 1083 : std::move(arrow_properties), writer);
961 : }
962 :
963 : /************************************************************************/
964 : /* CreateSchema() */
965 : /************************************************************************/
966 :
967 361 : void OGRParquetWriterLayer::CreateSchema()
968 : {
969 361 : CreateSchemaCommon();
970 361 : }
971 :
972 : /************************************************************************/
973 : /* CreateGeomField() */
974 : /************************************************************************/
975 :
976 27 : OGRErr OGRParquetWriterLayer::CreateGeomField(const OGRGeomFieldDefn *poField,
977 : int bApproxOK)
978 : {
979 27 : OGRErr eErr = OGRArrowWriterLayer::CreateGeomField(poField, bApproxOK);
980 53 : if (eErr == OGRERR_NONE &&
981 26 : m_aeGeomEncoding.back() == OGRArrowGeomEncoding::WKB
982 : #if ARROW_VERSION_MAJOR < 21
983 : // Geostatistics in Arrow 21 do not support geographic type for now
984 53 : && m_bEdgesSpherical
985 : #endif
986 : )
987 : {
988 0 : m_oWriterPropertiesBuilder.disable_statistics(
989 0 : parquet::schema::ColumnPath::FromDotString(
990 0 : m_poFeatureDefn
991 0 : ->GetGeomFieldDefn(m_poFeatureDefn->GetGeomFieldCount() - 1)
992 : ->GetNameRef()));
993 : }
994 27 : return eErr;
995 : }
996 :
997 : /************************************************************************/
998 : /* CreateWriter() */
999 : /************************************************************************/
1000 :
1001 361 : void OGRParquetWriterLayer::CreateWriter()
1002 : {
1003 361 : CPLAssert(m_poFileWriter == nullptr);
1004 :
1005 361 : if (m_poSchema == nullptr)
1006 : {
1007 43 : CreateSchema();
1008 : }
1009 : else
1010 : {
1011 318 : FinalizeSchema();
1012 : }
1013 :
1014 : auto arrowWriterProperties =
1015 361 : parquet::ArrowWriterProperties::Builder().store_schema()->build();
1016 1083 : CPL_IGNORE_RET_VAL(Open(*m_poSchema, m_poMemoryPool, m_poOutputStream,
1017 722 : m_oWriterPropertiesBuilder.build(),
1018 361 : std::move(arrowWriterProperties), &m_poFileWriter,
1019 : &m_poKeyValueMetadata));
1020 361 : }
1021 :
1022 : /************************************************************************/
1023 : /* ICreateFeature() */
1024 : /************************************************************************/
1025 :
1026 3390 : OGRErr OGRParquetWriterLayer::ICreateFeature(OGRFeature *poFeature)
1027 : {
1028 : // If not using SORT_BY_BBOX=YES layer creation option, we can directly
1029 : // write features to the final Parquet file
1030 3390 : if (!m_poTmpGPKGLayer)
1031 1186 : return OGRArrowWriterLayer::ICreateFeature(poFeature);
1032 :
1033 : // SORT_BY_BBOX=YES case: we write for now a serialized version of poFeature
1034 : // in a temporary GeoPackage file.
1035 :
1036 2204 : GIntBig nFID = poFeature->GetFID();
1037 2204 : if (!m_osFIDColumn.empty() && nFID == OGRNullFID)
1038 : {
1039 1102 : nFID = m_nTmpFeatureCount;
1040 1102 : poFeature->SetFID(nFID);
1041 : }
1042 2204 : ++m_nTmpFeatureCount;
1043 :
1044 4408 : std::vector<GByte> abyBuffer;
1045 : // Serialize the source feature as a single array of bytes to preserve it
1046 : // fully
1047 2204 : if (!poFeature->SerializeToBinary(abyBuffer))
1048 : {
1049 0 : return OGRERR_FAILURE;
1050 : }
1051 :
1052 : // SQLite3 limitation: a row must fit in slightly less than 1 GB.
1053 2204 : constexpr int SOME_MARGIN = 128;
1054 2204 : if (abyBuffer.size() > 1024 * 1024 * 1024 - SOME_MARGIN)
1055 : {
1056 0 : CPLError(CE_Failure, CPLE_NotSupported,
1057 : "Features larger than 1 GB are not supported");
1058 0 : return OGRERR_FAILURE;
1059 : }
1060 :
1061 4408 : OGRFeature oFeat(m_poTmpGPKGLayer->GetLayerDefn());
1062 2204 : oFeat.SetFID(nFID);
1063 2204 : oFeat.SetField(0, static_cast<int>(abyBuffer.size()), abyBuffer.data());
1064 2204 : const auto poSrcGeom = poFeature->GetGeometryRef();
1065 2204 : if (poSrcGeom && !poSrcGeom->IsEmpty())
1066 : {
1067 : // For the purpose of building an RTree, just use the bounding box of
1068 : // the geometry as the geometry.
1069 1202 : OGREnvelope sEnvelope;
1070 1202 : poSrcGeom->getEnvelope(&sEnvelope);
1071 2404 : auto poPoly = std::make_unique<OGRPolygon>();
1072 2404 : auto poLR = std::make_unique<OGRLinearRing>();
1073 1202 : poLR->addPoint(sEnvelope.MinX, sEnvelope.MinY);
1074 1202 : poLR->addPoint(sEnvelope.MinX, sEnvelope.MaxY);
1075 1202 : poLR->addPoint(sEnvelope.MaxX, sEnvelope.MaxY);
1076 1202 : poLR->addPoint(sEnvelope.MaxX, sEnvelope.MinY);
1077 1202 : poLR->addPoint(sEnvelope.MinX, sEnvelope.MinY);
1078 1202 : poPoly->addRingDirectly(poLR.release());
1079 1202 : oFeat.SetGeometryDirectly(poPoly.release());
1080 : }
1081 2204 : return m_poTmpGPKGLayer->CreateFeature(&oFeat);
1082 : }
1083 :
1084 : /************************************************************************/
1085 : /* FlushGroup() */
1086 : /************************************************************************/
1087 :
1088 333 : bool OGRParquetWriterLayer::FlushGroup()
1089 : {
1090 : #if PARQUET_VERSION_MAJOR >= 20
1091 : auto status = m_poFileWriter->NewRowGroup();
1092 : #else
1093 666 : auto status = m_poFileWriter->NewRowGroup(m_apoBuilders[0]->length());
1094 : #endif
1095 333 : if (!status.ok())
1096 : {
1097 0 : CPLError(CE_Failure, CPLE_AppDefined, "NewRowGroup() failed with %s",
1098 0 : status.message().c_str());
1099 0 : ClearArrayBuilers();
1100 0 : return false;
1101 : }
1102 :
1103 333 : auto ret = WriteArrays(
1104 1291 : [this](const std::shared_ptr<arrow::Field> &field,
1105 1291 : const std::shared_ptr<arrow::Array> &array)
1106 : {
1107 2582 : auto l_status = m_poFileWriter->WriteColumnChunk(*array);
1108 1291 : if (!l_status.ok())
1109 : {
1110 0 : CPLError(CE_Failure, CPLE_AppDefined,
1111 : "WriteColumnChunk() failed for field %s: %s",
1112 0 : field->name().c_str(), l_status.message().c_str());
1113 0 : return false;
1114 : }
1115 1291 : return true;
1116 : });
1117 :
1118 333 : ClearArrayBuilers();
1119 333 : return ret;
1120 : }
1121 :
1122 : /************************************************************************/
1123 : /* FixupWKBGeometryBeforeWriting() */
1124 : /************************************************************************/
1125 :
1126 51 : void OGRParquetWriterLayer::FixupWKBGeometryBeforeWriting(GByte *pabyWkb,
1127 : size_t nLen)
1128 : {
1129 51 : if (!m_bForceCounterClockwiseOrientation)
1130 0 : return;
1131 :
1132 51 : OGRWKBFixupCounterClockWiseExternalRing(pabyWkb, nLen);
1133 : }
1134 :
1135 : /************************************************************************/
1136 : /* FixupGeometryBeforeWriting() */
1137 : /************************************************************************/
1138 :
1139 1429 : void OGRParquetWriterLayer::FixupGeometryBeforeWriting(OGRGeometry *poGeom)
1140 : {
1141 1429 : if (!m_bForceCounterClockwiseOrientation)
1142 3 : return;
1143 :
1144 1426 : const auto eFlattenType = wkbFlatten(poGeom->getGeometryType());
1145 : // Polygon rings MUST follow the right-hand rule for orientation
1146 : // (counterclockwise external rings, clockwise internal rings)
1147 1426 : if (eFlattenType == wkbPolygon)
1148 : {
1149 74 : bool bFirstRing = true;
1150 151 : for (auto poRing : poGeom->toPolygon())
1151 : {
1152 85 : if ((bFirstRing && poRing->isClockwise()) ||
1153 8 : (!bFirstRing && !poRing->isClockwise()))
1154 : {
1155 72 : poRing->reversePoints();
1156 : }
1157 77 : bFirstRing = false;
1158 : }
1159 : }
1160 1352 : else if (eFlattenType == wkbMultiPolygon ||
1161 : eFlattenType == wkbGeometryCollection)
1162 : {
1163 135 : for (auto poSubGeom : poGeom->toGeometryCollection())
1164 : {
1165 71 : FixupGeometryBeforeWriting(poSubGeom);
1166 : }
1167 : }
1168 : }
1169 :
1170 : /************************************************************************/
1171 : /* WriteArrowBatch() */
1172 : /************************************************************************/
1173 :
1174 : #if PARQUET_VERSION_MAJOR > 10
1175 : inline bool
1176 25 : OGRParquetWriterLayer::WriteArrowBatch(const struct ArrowSchema *schema,
1177 : struct ArrowArray *array,
1178 : CSLConstList papszOptions)
1179 : {
1180 25 : if (m_poTmpGPKGLayer)
1181 : {
1182 : // When using SORT_BY_BBOX=YES option, we can't directly write the
1183 : // input array, because we need to sort features. Hence we fallback
1184 : // to the OGRLayer base implementation, which will ultimately call
1185 : // OGRParquetWriterLayer::ICreateFeature()
1186 0 : return OGRLayer::WriteArrowBatch(schema, array, papszOptions);
1187 : }
1188 :
1189 50 : return WriteArrowBatchInternal(
1190 : schema, array, papszOptions,
1191 50 : [this](const std::shared_ptr<arrow::RecordBatch> &poBatch)
1192 : {
1193 50 : auto status = m_poFileWriter->NewBufferedRowGroup();
1194 25 : if (!status.ok())
1195 : {
1196 0 : CPLError(CE_Failure, CPLE_AppDefined,
1197 : "NewBufferedRowGroup() failed with %s",
1198 0 : status.message().c_str());
1199 0 : return false;
1200 : }
1201 :
1202 25 : status = m_poFileWriter->WriteRecordBatch(*poBatch);
1203 25 : if (!status.ok())
1204 : {
1205 0 : CPLError(CE_Failure, CPLE_AppDefined,
1206 : "WriteRecordBatch() failed: %s",
1207 0 : status.message().c_str());
1208 0 : return false;
1209 : }
1210 :
1211 25 : return true;
1212 25 : });
1213 : }
1214 : #endif
1215 :
1216 : /************************************************************************/
1217 : /* TestCapability() */
1218 : /************************************************************************/
1219 :
1220 694 : inline bool OGRParquetWriterLayer::TestCapability(const char *pszCap) const
1221 : {
1222 : #if PARQUET_VERSION_MAJOR <= 10
1223 : if (EQUAL(pszCap, OLCFastWriteArrowBatch))
1224 : return false;
1225 : #endif
1226 :
1227 694 : if (m_poTmpGPKGLayer && EQUAL(pszCap, OLCFastWriteArrowBatch))
1228 : {
1229 : // When using SORT_BY_BBOX=YES option, we can't directly write the
1230 : // input array, because we need to sort features. So this is not
1231 : // fast
1232 1 : return false;
1233 : }
1234 :
1235 693 : return OGRArrowWriterLayer::TestCapability(pszCap);
1236 : }
1237 :
1238 : /************************************************************************/
1239 : /* CreateFieldFromArrowSchema() */
1240 : /************************************************************************/
1241 :
1242 : #if PARQUET_VERSION_MAJOR > 10
1243 479 : bool OGRParquetWriterLayer::CreateFieldFromArrowSchema(
1244 : const struct ArrowSchema *schema, CSLConstList papszOptions)
1245 : {
1246 479 : if (m_poTmpGPKGLayer)
1247 : {
1248 : // When using SORT_BY_BBOX=YES option, we can't directly write the
1249 : // input array, because we need to sort features. But this process
1250 : // only supports the base Arrow types supported by
1251 : // OGRLayer::WriteArrowBatch()
1252 0 : return OGRLayer::CreateFieldFromArrowSchema(schema, papszOptions);
1253 : }
1254 :
1255 479 : return OGRArrowWriterLayer::CreateFieldFromArrowSchema(schema,
1256 479 : papszOptions);
1257 : }
1258 : #endif
1259 :
1260 : /************************************************************************/
1261 : /* IsArrowSchemaSupported() */
1262 : /************************************************************************/
1263 :
1264 : #if PARQUET_VERSION_MAJOR > 10
1265 1095 : bool OGRParquetWriterLayer::IsArrowSchemaSupported(
1266 : const struct ArrowSchema *schema, CSLConstList papszOptions,
1267 : std::string &osErrorMsg) const
1268 : {
1269 1095 : if (m_poTmpGPKGLayer)
1270 : {
1271 : // When using SORT_BY_BBOX=YES option, we can't directly write the
1272 : // input array, because we need to sort features. But this process
1273 : // only supports the base Arrow types supported by
1274 : // OGRLayer::WriteArrowBatch()
1275 0 : return OGRLayer::IsArrowSchemaSupported(schema, papszOptions,
1276 0 : osErrorMsg);
1277 : }
1278 :
1279 1095 : if (schema->format[0] == 'e' && schema->format[1] == 0)
1280 : {
1281 1 : osErrorMsg = "float16 not supported";
1282 1 : return false;
1283 : }
1284 1094 : if (schema->format[0] == 'v' && schema->format[1] == 'u')
1285 : {
1286 1 : osErrorMsg = "StringView not supported";
1287 1 : return false;
1288 : }
1289 1093 : if (schema->format[0] == 'v' && schema->format[1] == 'z')
1290 : {
1291 1 : osErrorMsg = "BinaryView not supported";
1292 1 : return false;
1293 : }
1294 1092 : if (schema->format[0] == '+' && schema->format[1] == 'v')
1295 : {
1296 0 : if (schema->format[2] == 'l')
1297 : {
1298 0 : osErrorMsg = "ListView not supported";
1299 0 : return false;
1300 : }
1301 0 : else if (schema->format[2] == 'L')
1302 : {
1303 0 : osErrorMsg = "LargeListView not supported";
1304 0 : return false;
1305 : }
1306 : }
1307 2169 : for (int64_t i = 0; i < schema->n_children; ++i)
1308 : {
1309 1080 : if (!IsArrowSchemaSupported(schema->children[i], papszOptions,
1310 : osErrorMsg))
1311 : {
1312 3 : return false;
1313 : }
1314 : }
1315 1089 : return true;
1316 : }
1317 : #endif
1318 :
1319 : /************************************************************************/
1320 : /* SetMetadata() */
1321 : /************************************************************************/
1322 :
1323 13 : CPLErr OGRParquetWriterLayer::SetMetadata(CSLConstList papszMetadata,
1324 : const char *pszDomain)
1325 : {
1326 13 : if (!pszDomain || !EQUAL(pszDomain, "SHAPEFILE"))
1327 : {
1328 8 : return OGRLayer::SetMetadata(papszMetadata, pszDomain);
1329 : }
1330 5 : return CE_None;
1331 : }
1332 :
1333 : /************************************************************************/
1334 : /* GetDataset() */
1335 : /************************************************************************/
1336 :
1337 26 : GDALDataset *OGRParquetWriterLayer::GetDataset()
1338 : {
1339 26 : return m_poDataset;
1340 : }
|
