3937

Build # 3937

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/34d018257555d8f640a5a1644a85e60182991c05">34d018257<a href="https://github.com/apache/carbondata/commit/34d018257555d8f640a5a1644a85e60182991c05">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/1cd77223d96fb7d15da6a3e67092b3a8ae49a1d0">1cd77223d</a>">1cd77223d</a><a href="https://github.com/apache/carbondata/commit/34d018257555d8f640a5a1644a85e60182991c05"> into 188e7e4d4">188e7e4d4</a>

Run Details

69 of 69 new or added lines in 25 files covered. (100.0%)

64957 of 83598 relevant lines covered (77.7%)

1.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.73

/store/sdk/src/main/java/org/apache/carbondata/sdk/file/arrow/ArrowConverter.java

/*
 * 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.
 */

package org.apache.carbondata.sdk.file.arrow;

import java.io.IOException;
import java.nio.channels.Channels;
import java.util.TimeZone;

import org.apache.carbondata.core.stream.ExtendedByteArrayOutputStream;
import org.apache.carbondata.sdk.file.Schema;

import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.VectorUnloader;
import org.apache.arrow.vector.ipc.ArrowFileReader;
import org.apache.arrow.vector.ipc.ArrowFileWriter;
import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel;

public class ArrowConverter {

  private final BufferAllocator allocator;
  private VectorSchemaRoot root;
  private ArrowWriter arrowWriter;
  private org.apache.arrow.vector.types.pojo.Schema arrowSchema;
  private ExtendedByteArrayOutputStream out;
  private ArrowFileWriter writer;

  public ArrowConverter(Schema schema, int initialSize) {
    this.arrowSchema = ArrowUtils.toArrowSchema(schema, TimeZone.getDefault().getID());
    this.allocator =
        ArrowUtils.rootAllocator.newChildAllocator("toArrowBuffer", initialSize, Long.MAX_VALUE);
    this.root = VectorSchemaRoot.create(arrowSchema, allocator);
    this.arrowWriter = ArrowWriter.create(root);
    // currently blocklet level read and set initial value to 32 MB.
    this.out = new ExtendedByteArrayOutputStream(32 * 1024 * 1024);
    this.writer = new ArrowFileWriter(root, null, Channels.newChannel(out));
  }

  /**
   * write batch of row objects to Arrow vectors
   *
   * @param data
   */
  public void addToArrowBuffer(Object[] data) {
    int i = 0;
    while (i < data.length) {
      arrowWriter.write((Object[]) data[i]);
      i += 1;
    }
  }

  /**
   * To serialize arrow vectors to byte[]
   *
   * @return
   * @throws IOException
   */
  public byte[] toSerializeArray() throws IOException {
    arrowWriter.finish();
    writer.writeBatch();
    writer.close();
    arrowWriter.reset();
    root.close();
    byte[] bytes = out.toByteArray();
    allocator.close();
    out.close();
    return bytes;
  }

  /**
   * To copy arrow vectors to unsafe memory
   *
   * @return
   * @throws IOException
   */
  public long copySerializeArrayToOffHeap() throws IOException {
    arrowWriter.finish();
    writer.writeBatch();
    writer.close();
    arrowWriter.reset();
    root.close();
    long address = out.copyToAddress();
    allocator.close();
    out.close();
    return address;
  }

  /**
   * Utility API to convert back the arrow byte[] to arrow ArrowRecordBatch.
   * User need to close the ArrowRecordBatch after usage by calling ArrowRecordBatch.close()
   *
   * @param batchBytes input byte array
   * @param bufferAllocator arrow buffer allocator
   * @return ArrowRecordBatch
   * @throws IOException
   */
  public static ArrowRecordBatch byteArrayToArrowBatch(byte[] batchBytes,
      BufferAllocator bufferAllocator)
      throws IOException {
    ByteArrayReadableSeekableByteChannel in = new ByteArrayReadableSeekableByteChannel(batchBytes);
    ArrowFileReader reader = new ArrowFileReader(in, bufferAllocator);
    try {
      VectorSchemaRoot root = reader.getVectorSchemaRoot();
      VectorUnloader unloader = new VectorUnloader(root);
      reader.loadNextBatch();
      return unloader.getRecordBatch();
    } finally {
      reader.close();
    }
  }

  /**
   * To get the arrow vectors directly after filling from carbondata
   *
   * @return Arrow VectorSchemaRoot. which contains array of arrow vectors.
   */
  public VectorSchemaRoot getArrowVectors() throws IOException {
    arrowWriter.finish();
    writer.writeBatch();
    writer.close();
    return root;
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.sdk.file.arrow;
19
20	import java.io.IOException;
21	import java.nio.channels.Channels;
22	import java.util.TimeZone;
23
24	import org.apache.carbondata.core.stream.ExtendedByteArrayOutputStream;
25	import org.apache.carbondata.sdk.file.Schema;
26
27	import org.apache.arrow.memory.BufferAllocator;
28	import org.apache.arrow.vector.VectorSchemaRoot;
29	import org.apache.arrow.vector.VectorUnloader;
30	import org.apache.arrow.vector.ipc.ArrowFileReader;
31	import org.apache.arrow.vector.ipc.ArrowFileWriter;
32	import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
33	import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel;
34
35	public class ArrowConverter {
36
37	private final BufferAllocator allocator;
38	private VectorSchemaRoot root;
39	private ArrowWriter arrowWriter;
40	private org.apache.arrow.vector.types.pojo.Schema arrowSchema;
41	private ExtendedByteArrayOutputStream out;
42	private ArrowFileWriter writer;
43
44	public ArrowConverter(Schema schema, int initialSize) {	2✔
45	this.arrowSchema = ArrowUtils.toArrowSchema(schema, TimeZone.getDefault().getID());	2✔
46	this.allocator =	2✔
47	ArrowUtils.rootAllocator.newChildAllocator("toArrowBuffer", initialSize, Long.MAX_VALUE);	2✔
48	this.root = VectorSchemaRoot.create(arrowSchema, allocator);	2✔
49	this.arrowWriter = ArrowWriter.create(root);	2✔
50	// currently blocklet level read and set initial value to 32 MB.
51	this.out = new ExtendedByteArrayOutputStream(32 * 1024 * 1024);	2✔
52	this.writer = new ArrowFileWriter(root, null, Channels.newChannel(out));	2✔
53	}	2✔
54
55	/**
56	* write batch of row objects to Arrow vectors
57	*
58	* @param data
59	*/
60	public void addToArrowBuffer(Object[] data) {
61	int i = 0;	2✔
62	while (i < data.length) {	2✔
63	arrowWriter.write((Object[]) data[i]);	2✔
64	i += 1;	2✔
65	}
66	}	2✔
67
68	/**
69	* To serialize arrow vectors to byte[]
70	*
71	* @return
72	* @throws IOException
73	*/
74	public byte[] toSerializeArray() throws IOException {
75	arrowWriter.finish();	2✔
76	writer.writeBatch();	2✔
77	writer.close();	2✔
78	arrowWriter.reset();	2✔
79	root.close();	2✔
80	byte[] bytes = out.toByteArray();	2✔
81	allocator.close();	2✔
82	out.close();	2✔
83	return bytes;	2✔
84	}
85
86	/**
87	* To copy arrow vectors to unsafe memory
88	*
89	* @return
90	* @throws IOException
91	*/
92	public long copySerializeArrayToOffHeap() throws IOException {
93	arrowWriter.finish();	2✔
94	writer.writeBatch();	2✔
95	writer.close();	2✔
96	arrowWriter.reset();	2✔
97	root.close();	2✔
98	long address = out.copyToAddress();	2✔
99	allocator.close();	2✔
100	out.close();	2✔
101	return address;	2✔
102	}
103
104	/**
105	* Utility API to convert back the arrow byte[] to arrow ArrowRecordBatch.
106	* User need to close the ArrowRecordBatch after usage by calling ArrowRecordBatch.close()
107	*
108	* @param batchBytes input byte array
109	* @param bufferAllocator arrow buffer allocator
110	* @return ArrowRecordBatch
111	* @throws IOException
112	*/
113	public static ArrowRecordBatch byteArrayToArrowBatch(byte[] batchBytes,
114	BufferAllocator bufferAllocator)
115	throws IOException {
116	ByteArrayReadableSeekableByteChannel in = new ByteArrayReadableSeekableByteChannel(batchBytes);	2✔
117	ArrowFileReader reader = new ArrowFileReader(in, bufferAllocator);	2✔
118	try {
119	VectorSchemaRoot root = reader.getVectorSchemaRoot();	2✔
120	VectorUnloader unloader = new VectorUnloader(root);	2✔
121	reader.loadNextBatch();	2✔
122	return unloader.getRecordBatch();	2✔
123	} finally {
124	reader.close();	2✔
125	}	×
126	}
127
128	/**
129	* To get the arrow vectors directly after filling from carbondata
130	*
131	* @return Arrow VectorSchemaRoot. which contains array of arrow vectors.
132	*/
133	public VectorSchemaRoot getArrowVectors() throws IOException {
134	arrowWriter.finish();	2✔
135	writer.writeBatch();	2✔
136	writer.close();	2✔
137	return root;	2✔
138	}
139	}

apache / carbondata / 3937

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous