3731

Build # 3731

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/71e8d8e64ccb78d22b706ee50318371e81e055f4">71e8d8e64<a href="https://github.com/apache/carbondata/commit/71e8d8e64ccb78d22b706ee50318371e81e055f4">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/cbe553c39f9dd857b60a5cfc48ff2bdd8e871154">cbe553c39</a>">cbe553c39</a><a href="https://github.com/apache/carbondata/commit/71e8d8e64ccb78d22b706ee50318371e81e055f4"> into c1c50cf4b">c1c50cf4b</a>

Run Details

107 of 107 new or added lines in 8 files covered. (100.0%)

64088 of 82862 relevant lines covered (77.34%)

1.01 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.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
   * @param bufferAllocator
   * @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
   */
  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.sdk.file.Schema;
25
26	import org.apache.arrow.memory.BufferAllocator;
27	import org.apache.arrow.vector.VectorSchemaRoot;
28	import org.apache.arrow.vector.VectorUnloader;
29	import org.apache.arrow.vector.ipc.ArrowFileReader;
30	import org.apache.arrow.vector.ipc.ArrowFileWriter;
31	import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
32	import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel;
33
34	public class ArrowConverter {
35
36	private final BufferAllocator allocator;
37	private VectorSchemaRoot root;
38	private ArrowWriter arrowWriter;
39	private org.apache.arrow.vector.types.pojo.Schema arrowSchema;
40	private ExtendedByteArrayOutputStream out;
41	private ArrowFileWriter writer;
42
43	public ArrowConverter(Schema schema, int initialSize) {	2✔
44	this.arrowSchema = ArrowUtils.toArrowSchema(schema, TimeZone.getDefault().getID());	2✔
45	this.allocator =	2✔
46	ArrowUtils.rootAllocator.newChildAllocator("toArrowBuffer", initialSize, Long.MAX_VALUE);	2✔
47	this.root = VectorSchemaRoot.create(arrowSchema, allocator);	2✔
48	this.arrowWriter = ArrowWriter.create(root);	2✔
49	// currently blocklet level read and set initial value to 32 MB.
50	this.out = new ExtendedByteArrayOutputStream(32 * 1024 * 1024);	2✔
51	this.writer = new ArrowFileWriter(root, null, Channels.newChannel(out));	2✔
52	}	2✔
53
54	/**
55	* write batch of row objects to Arrow vectors
56	*
57	* @param data
58	*/
59	public void addToArrowBuffer(Object[] data) {
60	int i = 0;	2✔
61	while (i < data.length) {	2✔
62	arrowWriter.write((Object[]) data[i]);	2✔
63	i += 1;	2✔
64	}
65	}	2✔
66
67	/**
68	* To serialize arrow vectors to byte[]
69	*
70	* @return
71	* @throws IOException
72	*/
73	public byte[] toSerializeArray() throws IOException {
74	arrowWriter.finish();	2✔
75	writer.writeBatch();	2✔
76	writer.close();	2✔
77	arrowWriter.reset();	2✔
78	root.close();	2✔
79	byte[] bytes = out.toByteArray();	2✔
80	allocator.close();	2✔
81	out.close();	2✔
82	return bytes;	2✔
83	}
84
85	/**
86	* To copy arrow vectors to unsafe memory
87	*
88	* @return
89	* @throws IOException
90	*/
91	public long copySerializeArrayToOffHeap() throws IOException {
92	arrowWriter.finish();	2✔
93	writer.writeBatch();	2✔
94	writer.close();	2✔
95	arrowWriter.reset();	2✔
96	root.close();	2✔
97	long address = out.copyToAddress();	2✔
98	allocator.close();	2✔
99	out.close();	2✔
100	return address;	2✔
101	}
102
103	/**
104	* Utility API to convert back the arrow byte[] to arrow ArrowRecordBatch.
105	* User need to close the ArrowRecordBatch after usage by calling ArrowRecordBatch.close()
106	*
107	* @param batchBytes
108	* @param bufferAllocator
109	* @return ArrowRecordBatch
110	* @throws IOException
111	*/
112	public static ArrowRecordBatch byteArrayToArrowBatch(byte[] batchBytes,
113	BufferAllocator bufferAllocator)
114	throws IOException {
115	ByteArrayReadableSeekableByteChannel in = new ByteArrayReadableSeekableByteChannel(batchBytes);	2✔
116	ArrowFileReader reader = new ArrowFileReader(in, bufferAllocator);	2✔
117	try {
118	VectorSchemaRoot root = reader.getVectorSchemaRoot();	2✔
119	VectorUnloader unloader = new VectorUnloader(root);	2✔
120	reader.loadNextBatch();	2✔
121	return unloader.getRecordBatch();	2✔
122	} finally {
123	reader.close();	2✔
124	}	×
125	}
126
127	/**
128	* To get the arrow vectors directly after filling from carbondata
129	*
130	* @return
131	*/
132	public VectorSchemaRoot getArrowVectors() throws IOException {
133	arrowWriter.finish();	2✔
134	writer.writeBatch();	2✔
135	writer.close();	2✔
136	return root;	2✔
137	}
138	}

apache / carbondata / 3731

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